实战：Deep Research Agent

# uv add ddgs httpx
# 注：ddgs 的旧包名是 duckduckgo-search，若安装失败可回退旧名
from ddgs import DDGS
import httpx
import re
from pathlib import Path
from pydantic import BaseModel, Field

# ---------- web_search ----------
class WebSearchArgs(BaseModel):
    query: str = Field(description="搜索关键词。精炼具体，一次只搜一个主题")
    max_results: int = Field(default=5, description="返回结果数，默认 5")

def web_search(query: str, max_results: int = 5) -> str:
    try:
        results = DDGS().text(query, max_results=max_results)
    except Exception as e:
        return f"搜索失败：{e}"
    if not results:
        return "没有搜索到任何结果，建议更换关键词"
    lines = []
    for i, r in enumerate(results, 1):
        lines.append(f"[{i}] {r['title']}\n    URL: {r['href']}\n    摘要: {r['body']}")
    return "\n".join(lines)

# ---------- read_webpage ----------
class ReadWebpageArgs(BaseModel):
    url: str = Field(description="要阅读的网页 URL（来自搜索结果的 URL 字段）")
    max_chars: int = Field(default=3000, description="最多返回的字符数")

def read_webpage(url: str, max_chars: int = 3000) -> str:
    try:
        r = httpx.get(url, timeout=15, follow_redirects=True,
                      headers={"User-Agent": "Mozilla/5.0"})
        html = r.text
        html = re.sub(r"<(script|style)[^>]*>.*?</\1>", "", html,
                      flags=re.S | re.I)          # 去掉脚本和样式
        text = re.sub(r"<[^>]+>", " ", html)        # 去标签
        text = re.sub(r"\s+", " ", text).strip()    # 压缩空白
        if len(text) > max_chars:
            return text[:max_chars] + f"...（已截断，全文 {len(text)} 字符）"
        return text or "（页面无可提取文本）"
    except Exception as e:
        return f"读取失败：{e}"

registry = Registry()
registry.register(Tool(web_search, WebSearchArgs,
    "联网搜索最新信息。当问题涉及时事、数据、你不确定的事实时必须先调用。"
    "返回带编号的结果列表。"))
registry.register(Tool(read_webpage, ReadWebpageArgs,
    "读取网页正文内容。需要深入阅读某个搜索结果的原文时使用。"))

SYSTEM = """你是一个严谨的研究助手，只依据搜索到的信息回答问题。

## 行为规范
1. 回答事实性问题前，必须先调用 web_search；信息不足时用 read_webpage 阅读原文
2. 搜索结果矛盾时，优先采信原文，并说明分歧
3. 确实查不到时，明确说"未找到可靠来源"，禁止编造

## 输出格式
- 每条事实性结论后标注来源编号，如：某产品 2025 年销量为 X 万台[1]
- 回答末尾列出"参考来源"，格式：[n] 标题 — URL
- 回答保持精炼，与问题无关的信息不写"""

agent = Agent(system=SYSTEM, registry=registry, max_rounds=6)

def logger(event, data):
    if event == Ev.TOOL_CALL:
        print(f"  ⚙ {data['call'][0]}({data['call'][1][:70]})")
    elif event == Ev.TOOL_RESULT:
        print(f"  ↳ {data['result'][:60]}")
    elif event == Ev.FINISH:
        print(f"  ✔ {data['round']} 轮完成")

agent.on(logger)
answer = agent.run("Qwen3 发布于什么时候？和 DeepSeek-V3 比主要特点是什么？")
print(answer)

  ⚙ web_search({"query": "Qwen3 发布时间 特点"})
  ↳ [1] Qwen3 发布：混合推理模式... URL: https://... 摘要: ...
  ⚙ web_search({"query": "DeepSeek-V3 特点"})
  ↳ [1] DeepSeek-V3 技术报告... 摘要: ...
  ⚙ read_webpage({"url": "https://...(第 1 条结果的 URL)"})
  ↳ Qwen3 于 2025 年 4 月发布，提供 0.6B~235B 八种规模...
  ✔ 4 轮完成