1. 引言：为什么从 Search 走向 Extract

在构建 AI Agent 的过程中，一个核心问题始终绕不开：Agent 如何获取外部世界的实时信息？早期方案普遍依赖 Search——让大模型调用搜索接口，把返回的网页片段塞进上下文。但随着业务深入，Search 的局限性越来越明显：搜索结果噪声大、信息密度低、Token 消耗高，而且往往拿不到页面中真正关键的结构化数据。

于是，业界开始转向 Extract 范式：不是让 Agent 去“搜”，而是让 Agent 明确告诉工具“我要从哪个页面、提取哪些字段”，由专门的抽取引擎负责解析、清洗和结构化输出。本文将从架构角度拆解这一转变，并给出可运行的代码实战。

2. Search 与 Extract 的本质区别

Search 和 Extract 并不是互斥的，而是处于同一信息获取链路的不同阶段。理解它们的差异，是设计 Agent 工具链的第一步。

维度	Search（搜索）	Extract（抽取）
输入	关键词或自然语言查询	目标 URL + 结构化字段定义
输出	网页链接列表 + 摘要片段	JSON 结构化数据
信息密度	低，噪声多	高，按需取用
Token 消耗	高，常需塞入整页内容	低，只保留目标字段
适用场景	发现信息、探索未知	已知来源、重复采集
失败模式	结果不相关、排名靠后	选择器失效、页面结构变化

一个典型的 Agent 工作流是：先用 Search 发现候选页面，再用 Extract 从选定页面中精确取出所需数据。两者配合，才能兼顾探索能力和执行效率。

3. 整体架构设计

一个面向 AI Agent 的网页工具架构，通常包含以下核心模块：

flowchart TD
    A[Agent / LLM] --> B[Tool Router]
    B --> C[Search Tool]
    B --> D[Extract Tool]
    C --> E[Search API / Index]
    D --> F[Fetch & Render]
    F --> G[Content Cleaner]
    G --> H[Schema Extractor]
    H --> I[Structured JSON]
    I --> A

整个链路可以拆成四层：

• 接口层：以 OpenAI Function Calling 或 Anthropic Tool Use 的形式，把 Search 和 Extract 暴露给大模型。
• 调度层：Tool Router 根据用户意图和上下文，决定调用哪个工具、传什么参数。
• 执行层：Fetch 负责抓取页面，Content Cleaner 负责去噪，Schema Extractor 负责按字段定义抽取。
• 返回层：把结构化 JSON 回传给 Agent，由 LLM 组织成自然语言答案。

4. 环境准备与依赖

本文代码基于 Python 3.10+，核心依赖如下：

pip install httpx beautifulsoup4 lxml pydantic openai

各库的用途：

• httpx：异步 HTTP 客户端，负责抓取网页。
• beautifulsoup4 + lxml：解析 HTML，定位目标节点。
• pydantic：定义抽取字段的 Schema，并做输出校验。
• openai：接入 LLM，实现 Agent 的工具调用循环。

5. 实战一：实现一个最小可用的 Extract 工具

我们先从最核心的 Extract 工具开始。它的职责是：给定 URL 和字段定义，返回结构化 JSON。

5.1 定义抽取 Schema

使用 Pydantic 定义我们希望从页面中提取的字段。以提取一篇技术博客的作者、标题和正文摘要为例：

from pydantic import BaseModel, Field
from typing import Optional
class ArticleSchema(BaseModel):
title: str = Field(description="文章标题")
author: Optional[str] = Field(default=None, description="作者名称")
summary: Optional[str] = Field(default=None, description="文章摘要或导语")
publish_date: Optional[str] = Field(default=None, description="发布日期")

5.2 抓取与清洗

抓取页面后，第一步是去除 script、style、nav、footer 等噪声节点，保留正文主体：

import httpx
from bs4 import BeautifulSoup
async def fetch_and_clean(url: str) -> str:
async with httpx.AsyncClient(
headers={"User-Agent": "Mozilla/5.0 (compatible; AI-Agent/1.0)"},
timeout=15.0,
follow_redirects=True,
) as client:
resp = await client.get(url)
resp.raise_for_status()
soup = BeautifulSoup(resp.text, "lxml")
移除噪声节点
for tag in soup(["script", "style", "nav", "footer", "aside", "iframe"]):
tag.decompose()
优先取 article 或 main 区域，否则取 body
main = soup.find("article") or soup.find("main") or soup.body
return str(main) if main else ""

5.3 基于 CSS 选择器的字段抽取

对于结构稳定的页面，CSS 选择器是最快、最省 Token 的方式。我们把字段与选择器映射起来：

from bs4 import BeautifulSoup
SELECTORS = {
"title": "h1",
"author": ".author, .byline, meta[name=author]",
"summary": ".summary, .description, meta[name=description]",
"publish_date": "time, .date, meta[property=article:published_time]",
}
def extract_by_selectors(html: str, schema: dict) -> dict:
soup = BeautifulSoup(html, "lxml")
result = {}
for field, selector in schema.items():
node = soup.select_one(selector)
if node is None:
result[field] = None
continue
if node.name == "meta":
result[field] = node.get("content")
else:
result[field] = node.get_text(strip=True)
return result

5.4 用 LLM 做兜底抽取

选择器方案在页面改版时会失效。此时可以退回到 LLM 抽取：把清洗后的文本截断后交给模型，让它按 Schema 输出 JSON。这里用 OpenAI 的 JSON Mode 做演示：

import json
from openai import AsyncOpenAI
client = AsyncOpenAI()
async def extract_with_llm(html: str, schema_model) -> dict:
# 简单去标签，保留纯文本并截断
soup = BeautifulSoup(html, "lxml")
text = soup.get_text(" ", strip=True)[:6000]
resp = await client.chat.completions.create(
    model="gpt-4o-mini",
    response_format={"type": "json_object"},
    messages=[
        {
            "role": "system",
            "content": (
                "你是一个网页信息抽取器。请从用户提供的网页文本中"
                "提取指定字段，只输出 JSON。"
            ),
        },
        {
            "role": "user",
            "content": (
                f"字段定义：{schema_model.model_json_schema()}\n"
                f"网页文本：{text}"
            ),
        },
    ],
)
data = json.loads(resp.choices[0].message.content)
return schema_model(**data).model_dump()

5.5 组合成完整 Extract 工具

async def extract(url: str, schema_model, use_llm_fallback: bool = True) -> dict:
    html = await fetch_and_clean(url)
# 先尝试选择器方案
selectors = {
    "title": "h1",
    "author": ".author, .byline",
    "summary": ".summary, .description",
    "publish_date": "time, .date",
}
result = extract_by_selectors(html, selectors)
如果关键字段缺失，回退到 LLM
if use_llm_fallback and not result.get("title"):
result = await extract_with_llm(html, schema_model)
return result

6. 实战二：实现 Search 工具并串联 Agent

Search 工具负责“发现”。这里我们用一个可替换的搜索接口封装，实际生产中可以接入 SerpAPI、Bing Search API 或自建索引。

6.1 封装 Search 工具

import os
import httpx
async def search(query: str, top_k: int = 5) -> list[dict]:
"""调用搜索 API，返回候选页面列表。"""
api_key = os.environ["SEARCH_API_KEY"]
url = "https://api.search-provider.example/v1/search"
params = {"q": query, "num": top_k}
headers = {"Authorization": f"Bearer {api_key}"}
async with httpx.AsyncClient(timeout=10.0) as client:
    resp = await client.get(url, params=params, headers=headers)
    resp.raise_for_status()
items = resp.json().get("items", [])
return [
{"title": it["title"], "url": it["link"], "snippet": it.get("snippet", "")}
for it in items
]

6.2 用 Function Calling 暴露工具

为了让 LLM 能自主决定调用哪个工具，我们需要把工具定义传给模型：

tools = [
    {
        "type": "function",
        "function": {
            "name": "search",
            "description": "搜索网页，返回候选链接列表。用于发现信息。",
            "parameters": {
                "type": "object",
                "properties": {
                    "query": {"type": "string", "description": "搜索关键词"}
                },
                "required": ["query"],
            },
        },
    },
    {
        "type": "function",
        "function": {
            "name": "extract",
            "description": "从指定 URL 提取结构化字段。用于获取精确数据。",
            "parameters": {
                "type": "object",
                "properties": {
                    "url": {"type": "string", "description": "目标页面 URL"}
                },
                "required": ["url"],
            },
        },
    },
]

6.3 Agent 主循环

下面实现一个简单的 Agent 循环：模型先决定调用哪个工具，我们执行工具并把结果回传给模型，直到模型给出最终答案。

import json
from openai import AsyncOpenAI
client = AsyncOpenAI()
TOOL_IMPL = {
"search": search,
"extract": lambda url: extract(url, ArticleSchema),
}
async def run_agent(user_query: str) -> str:
messages = [{"role": "user", "content": user_query}]
for _ in range(5):  # 限制最大工具调用轮数
    resp = await client.chat.completions.create(
        model="gpt-4o",
        messages=messages,
        tools=tools,
        tool_choice="auto",
    )
    msg = resp.choices[0].message
    messages.append(msg)
if not msg.tool_calls:
    return msg.content or ""
for call in msg.tool_calls:
fn = call.function
args = json.loads(fn.arguments)
print(f"[Agent] 调用 {fn.name}: {args}")
result = await TOOL_IMPL[fn.name](**args)
messages.append({
    "role": "tool",
    "tool_call_id": call.id,
    "content": json.dumps(result, ensure_ascii=False),
})
return "达到最大工具调用轮数，未得到最终答案。"

6.4 运行示例

import asyncio
async def main():
answer = await run_agent("帮我找到一篇关于 RAG 的最新技术博客，并提取它的标题和作者。")
print(answer)
asyncio.run(main())

运行后，控制台会输出类似下面的工具调用日志：

[Agent] 调用 search: {'query': 'RAG 最新技术博客 2025'}
[Agent] 调用 extract: {'url': 'https://example.com/rag-blog'}

最终，模型会基于 Extract 返回的 JSON 组织成自然语言答案。

7. 架构演进：从单工具到工具编排

上面的示例已经能跑通，但生产环境还需要考虑几个关键演进方向。

7.1 缓存层

同一 URL 的抽取结果在短时间内往往不会变化。引入缓存可以显著降低成本：

import time
from functools import lru_cache
@lru_cache(maxsize=256)
def cached_extract(url: str, ttl_seconds: int = 3600):
# 实际项目中可用 Redis，这里用简单的时间戳模拟
return extract(url, ArticleSchema)

7.2 多级回退策略

抽取失败时，按“选择器 → LLM → 人工规则”的优先级逐级回退，能兼顾速度与鲁棒性。建议把每一级的耗时和成功率记录下来，用于持续优化选择器。

7.3 结构化输出校验

LLM 输出偶尔会不符合 Schema。用 Pydantic 校验并做一次“修复式重试”，可以显著提升成功率：

from pydantic import ValidationError
async def extract_with_retry(url: str, schema_model, max_retries: int = 2) -> dict:
for attempt in range(max_retries):
try:
data = await extract_with_llm(url, schema_model)
return schema_model(**data).model_dump()
except ValidationError as e:
if attempt == max_retries - 1:
raise
print(f"校验失败，重试中：{e}")

8. 踩坑与最佳实践

• 反爬与限流：抓取前先检查 robots.txt，控制请求频率，必要时使用代理池。
• 页面动态渲染：部分页面由 JavaScript 渲染，httpx 拿不到正文。此时需要接入 Playwright 或 Puppeteer 做无头浏览器渲染。
• 选择器脆弱性：优先使用稳定的语义化选择器（如 article h1），避免依赖频繁变化的 class 名。
• Token 预算：LLM 兜底抽取前务必截断文本，并设置最大输入长度，防止上下文溢出。
• 错误处理：网络超时、404、解析失败都要有明确的错误码返回给 Agent，让模型能据此调整策略。

9. 总结

从 Search 到 Extract 的转变，本质上是把“让模型读网页”变成“让工具替模型读网页”。Search 负责发现，Extract 负责精取，两者通过 Function Calling 无缝衔接，既保留了探索能力，又大幅提升了信息密度和执行效率。

本文给出的代码已经构成一个可运行的最小闭环。你可以在此基础上继续扩展：接入更多数据源、增加缓存与监控、把抽取结果写入知识库，逐步构建出真正生产可用的 AI Agent 网页工具链。