作者為 AI 智能體建立文件爬蟲,49 秒抓取 60 頁 Python 文件
I Gave My AI Agents Their Own Documentation Crawler, and Pulled 60 Pages of Clean Markdown in 49 Seconds
作者為 AI 智能體建立文件爬蟲,可按需抓取網頁文件並轉成 Markdown;該工具在 49 秒內抓取 60 頁 Python 文件。方案採用同一網域抓取與並行處理,並把轉換後的文件交由 Agent 作 RAG 式檢索;作者提醒不要執行爬取內容中的程式碼。
Introduction
My AI agents were flying blind. They could write code, but when they needed to reference library docs, they either hallucinated APIs or stalled waiting for me to paste documentation. I needed a way for them to fetch and convert web docs into clean Markdown on demand. So I built a lightweight crawler that runs inside the agent's tool loop. It pulled 60 pages of Python docs and converted them to Markdown in 49 seconds. Here's how it works.
The Problem
Agents are only as good as their context. When an agent needs to use a library like requests or pydantic, it often relies on stale training data. The result: wrong function signatures, deprecated parameters, and broken code. I wanted my agents to fetch current documentation, convert it to Markdown (which LLMs handle well), and use it as context. But existing crawlers are heavy, require external services, or output messy HTML.
The Solution
I built a minimal crawler that:
- Starts from a seed URL.
- Follows links within the same domain.
- Fetches pages concurrently.
- Converts HTML to Markdown using
html2text. - Respects
robots.txtand rate limits. - Stores results in a simple in-memory cache.
The agent calls it as a tool: crawl_docs(url, max_pages=60). The crawler returns a list of Markdown documents. The agent then uses those documents for RAG-style retrieval.
Implementation
Dependencies
pip install requests beautifulsoup4 html2text
The Crawler
import requests
from bs4 import BeautifulSoup
from urllib.parse import urljoin, urlparse
from concurrent.futures import ThreadPoolExecutor, as_completed
import html2text
import time
from collections import deque
class DocsCrawler:
def __init__(self, base_url, max_pages=60, max_workers=8, delay=0.1):
self.base_url = base_url
self.base_domain = urlparse(base_url).netloc
self.max_pages = max_pages
self.max_workers = max_workers
self.delay = delay
self.visited = set()
self.queue = deque([base_url])
self.results = []
self.session = requests.Session()
self.session.headers.update({
'User-Agent': 'DocsCrawler/1.0 (+https://example.com/bot)'
})
def is_valid(self, url):
parsed = urlparse(url)
return (
parsed.netloc == self.base_domain
and parsed.scheme in ('http', 'https')
and not parsed.path.endswith(('.pdf', '.zip', '.png', '.jpg'))
)
def fetch(self, url):
try:
time.sleep(self.delay)
resp = self.session.get(url, timeout=10)
resp.raise_for_status()
return resp.text
except Exception as e:
print(f"Failed {url}: {e}")
return None
def extract_links(self, html, current_url):
soup = BeautifulSoup(html, 'html.parser')
links = set()
for a in soup.find_all('a', href=True):
href = a['href']
full_url = urljoin(current_url, href)
if self.is_valid(full_url):
links.add(full_url)
return links
def html_to_markdown(self, html):
h = html2text.HTML2Text()
h.ignore_links = False
h.ignore_images = True
h.body_width = 0
return h.handle(html)
def crawl(self):
with ThreadPoolExecutor(max_workers=self.max_workers) as executor:
futures = {}
while self.queue and len(self.visited) < self.max_pages:
# Fill the executor with pending URLs
while self.queue and len(futures) < self.max_workers:
url = self.queue.popleft()
if url in self.visited:
continue
self.visited.add(url)
futures[executor.submit(self.fetch, url)] = url
# Process completed futures
for future in as_completed(futures):
url = futures.pop(future)
html = future.result()
if not html:
continue
md = self.html_to_markdown(html)
self.results.append({'url': url, 'markdown': md})
for link in self.extract_links(html, url):
if link not in self.visited:
self.queue.append(link)
if len(self.visited) >= self.max_pages:
break
return self.results
Usage
if __name__ == '__main__':
start = time.time()
crawler = DocsCrawler('https://docs.python.org/3/library/', max_pages=60)
docs = crawler.crawl()
elapsed = time.time() - start
print(f"Crawled {len(docs)} pages in {elapsed:.2f} seconds")
# Save to files
for i, doc in enumerate(docs):
with open(f'doc_{i}.md', 'w') as f:
f.write(f"# Source: {doc['url']}\n\n{doc['markdown']}")
Integrating with an Agent
The agent loop calls the crawler as a tool. Here's a simplified agent loop with explicit termination:
def agent_loop(task, max_iterations=10):
context = []
for i in range(max_iterations):
action = llm_decide(task, context)
if action['type'] == 'crawl_docs':
docs = crawler.crawl()
context.extend(docs)
elif action['type'] == 'answer':
return action['content']
else:
context.append(f"Unknown action: {action}")
return "Max iterations reached"
Security note: Never execute arbitrary code from crawled pages. The crawler only extracts text. If you use eval or exec on Markdown content, you open yourself to code injection. Always sanitize and treat crawled content as untrusted input.
Key Takeaways
- Concurrency matters: Using a thread pool cut crawl time from minutes to seconds.
-
Markdown is ideal for LLMs:
html2textproduces clean, token-efficient text. -
Respect boundaries: Filter by domain, skip binary files, and honor
robots.txt(not shown for brevity but recommended). - Agent integration is simple: Expose the crawler as a tool; the agent decides when to use it.
-
Security first: Crawled content is untrusted. Never
evalit. - Performance: 60 pages in 49 seconds on a laptop with 8 workers. Your mileage may vary with network and site speed.
The full code is on GitHub. Happy crawling!
來源:DEV Community · dev.to