LangGraph 智能体策略(~95% SimpleQA 的那条)
30 秒导读: 这是 LDR 招牌的「自主 agent」检索策略。它不像流水线那样按固定步数迭代,而是把一个基于 LangChain
create_agent的工具调用 agent 放出去,让 LLM 自己决定搜什么词、用哪个专用引擎、要不要抓全文、要不要把难题拆成子问题丢给并行子 agent、以及何时收尾合成。招牌 指标(SimpleQA ~95%)就是这条策略跑出来的。
本章聚焦三件事:agent 自主决策 + 子 agent 并行 + 工具化检索。两阶段检索机制见 02;引用合成与报告的通用部分见 04;默认的 source-based 流水线骨架见 01。
1. 这是什么(和流水线策略比,差在哪)
一句话定义: 一个让 LLM 用工具「自己找答案」的研究策略——你给它一个问题,它自主地搜、读、拆、合,直到觉得够了才写报告。
它解决什么问题。 流水线式策略(01 的默认 source-based)把研究写死成固定节拍:生成问题 → 搜索 → 过滤 → 再迭代 N 轮。这套稳,但不会随问题难度伸缩——简单事实题也走满流程,复杂多跳题又可能轮数不够。agentic 策略把「下一步做什么」这个决策从代码里挪到 LLM 里。
本质差异(这是本章的题眼)。 同样是「多轮检索」,两者谁在做决策完全不同:
| 维度 | 流水线策略(01) | LangGraph agent 策略(本章) |
|---|---|---|
| 下一步搜什么 | 代码按模板生成子问题 | LLM 看着已有结果自己想 |
| 用 哪个引擎 | 策略固定选定 | LLM 从工具清单里挑(web / arXiv / PubMed…) |
| 迭代几轮 | 固定 iterations(通常 1-5) | LLM 自己决定,上限是递归限而非轮数 |
| 何时抓全文 | 规则触发 | LLM 判断「snippet 不够」时调 fetch_content |
| 拆子问题 | 无 / 固定分解 | LLM 调 research_subtopic 派并行子 agent |
| 何时收尾 | 跑满轮数 | LLM「觉得够了」就直接写答案 |
用源码里的话说,它「替换掉 MCP 策略里那个手写的 ReAct 循环」——决策不再是 while 循环里的 if/else,而是 agent 每一步的一次 LLM 调用(langgraph_agent_strategy.py:476-482,类 docstring)。
一句话直觉。 流水线像自动化流水线:工位、顺序、节拍都排死。agent 像派一个实习研究员进图书馆:你只说「搞清楚这个问题」,他自己决定先查哪个数据库、要不要复印全文、要不要喊几个同事分头查,查够了回来交报告。本章讲的就是这个「实习研究员」的骨架、它的工具箱、以及它喊来的「同事」怎么并行还不打架。
为什么每个 iteration 需要更多轮。 对流水线,一次 iteration 是一整轮搜索;对 agent,一次 iteration 只是一个 LLM→工具的来回。所以两者的迭代常量量级差一个数(下面 §3 讲)。
2. 顶层全景(它大概怎么转)
怎么读这张图: 从 上到下是一次 analyze_topic 的生命周期;中间那圈虚线是 agent 的自主循环(LLM 决策 ↔ 工具执行,反复到收尾);右下角是被 research_subtopic 派出去的并行子 agent。所有搜索结果都往同一个 SearchResultsCollector 汇。
用户问题 query
│
▼
┌─────────────────────────────┐
│ LangGraphAgentStrategy │
│ .analyze_topic() │ ← 主入口
│ · 建工具箱 _build_tools │
│ · 建 system_prompt │
│ · create_agent(...) │
└──────────────┬───────── ─────┘
│ agent.stream(updates)
┌────────────▼─────────────┐
┆ 主 agent 自主循环 ┆ ← LLM 每步自己决定下一步
┆ ┌────────┐ ┌────────┐ ┆
┆ │ LLM │→ │ 工具 │ ┆
┆ │ 决策 │← │ 执行 │ ┆
┆ └────────┘ └───┬────┘ ┆
└──────────────────┼───────┘
│ 工具调用分四类
┌──────────┬─────────┼──────────┬────────────┐
▼ ▼ ▼ ▼ │
web_search search_* fetch_content research_subtopic
(主引擎) (专用引擎) (抓全文) (派并行子 agent)
│ │ │ │ │
│ │ │ ┌────▼────┬────┬──┴─┐
│ │ │ │子agent 1│... │≤4 │ 线程池
│ │ │ └────┬────┴────┴────┘
└──────────┴─────────┴──────────┴─────────────┘
│ 全部结果 add_results()
▼
┌───────────────────────────┐
│ SearchResultsCollector │ ← 线程安全,统一发引用序号
│ · _all_links(共享) │ 写入 all_links_of_system
└──────────────┬─ ───────────┘
▼
_finalize → CitationHandler → 带引用的报告
部件一句话职责:
| 部件 | 干什么 | 在哪 |
|---|---|---|
LangGraphAgentStrategy | 策略类,搭 agent、跑 stream、收尾 | strategies/langgraph_agent_strategy.py:476 |
analyze_topic | 主入口:建工具→建 prompt→建 agent→流式跑→合成 | 同上 :780-1126 |
_build_tools | 组装工具箱(web / 专用 / fetch / subtopic) | 同上 :619-776 |
SearchResultsCollector | 线程安全收集器,统一分配引用序号 | 同上 :60-136 |
research_subtopic 工具 | 把子问题分给并行子 agent | 同上 :291-468 |
build_fetch_tool | 造 fetch_content 抓全文工具 | tools/fetch/__init__.py:355 |
3. 整体构造:一个工具调用 agent 的骨架
这节讲 agent 本体怎么搭起来:用什么建、迭代常量怎么定、__init__ 装配了什么、analyze_topic 怎么把它跑起来。
3.1 用 create_agent 建工具调用 agent
策略不自己写 ReAct 循环,而是把 model + tools + system_prompt 交给 LangChain 的 create_agent,拿回一个能 .stream() 的 agent(langgraph_agent_strategy.py:895-900)。循环体(「调 LLM → 解析 tool_calls → 执行工具 → 回填结果 → 再调 LLM」)由 LangGraph 底层负责,策略只管装配和消费流。
一个已知坑(源码明写): create_agent 会把工具绑到基础 LLM 上(bind_tools 经 ProcessingLLMWrapper.__getattr__ 透传),绕过了 wrapper 的 <think> 标签剥离。所以推理模型在这条 agent 循环里的思考标签不会被剥掉——只是显示上的瑕疵,不会崩(:889-894 注释)。文档如实记下:这是已知限制,别靠「每次重新包一层」去修。
3.2 迭代常量:为什么是 50 不是 5
agent 每步只走一个 LLM→工具来回,所以需要的「轮数」比流水线多一个量级。常量集中在文件顶部:
| 常量 | 值 | 含义 | 位置 |
|---|---|---|---|
DEFAULT_MAX_ITERATIONS | 50 | agent 默认上限,「比流水线策 略多得多」 | :32-34 |
MIN_ITERATIONS | 10 | 低于这个数 agent「几乎干不了事」 | :35 |
MAX_SUBTOPICS | 8 | 一次最多拆几个子问题 | :37 |
MAX_SUBAGENT_WORKERS | 4 | 子 agent 线程池并行度 | :38 |
SUBAGENT_TIMEOUT_SECONDS | 1800 | 单个子 agent 超时(30 分钟) | :36 |
关键细节:递归限而非轮数封顶。 max_iterations 传进来后不是硬 clamp,而是「低于 MIN_ITERATIONS 就当没设,用默认 50」——避免把 agent 卡在一个没用的低值上(__init__ 里 :512-516)。真正传给 LangGraph 的是递归上限,换算公式是 effective_max * 2 + 1(:911-912),留出「LLM 节点 + 工具节点」交替的空间。
3.3 __init__:装配阶段做了什么
构造时确定四件事(:484-539):
- 模型与搜索引擎:
self.model、self.search,并从设置解析出主引擎名_resolve_engine_name()(读search.tool,拿不到就从类名启发式推,兜底duckduckgo,:541-551)。 - 迭代上限:按 3.2 的「低于最小值就用默认」规则定
self.max_iterations。 - 收集器:
self.collector = SearchResultsCollector(self.all_links_of_system)—— 把策略的共享链接表交给收集器(:524,详见 §4)。 - 抓取模式:读
search.fetch.mode,不在FETCH_MODES里就回退summary_focus_query(:526-536)。
3.4 analyze_topic:一次研究的主线
analyze_topic(query) 是主入口,骨架如下(:780-1126):
# 示意,非源码 —— 只画主干,省略进度上报/异常分支
def analyze_topic(self, query):
self.collector.reset() # 详细报告模式下每小节重置
tools = self._build_tools(overall_query=query) # 工具箱(§5)
system_prompt = build_prompt(fetch_mode, policy_addendum) # 含策略措辞
agent = create_agent(model, tools, system_prompt)
for chunk in agent.stream({"messages": [...]}, config, "updates"):
self.check_termination() # 允许外部中止
# 解析 chunk:agent 节点→上报"在搜什么";tools 节点→上报"读到啥"+心跳
return self._finalize(query, final_content, ...) # §7
几处值得点出的装配细节:
- system_prompt 随抓取模式改口径(
:807-822):disabled时告诉 agent「只能靠 snippet」;summary 模式让它「调fetch_content(url, focus)并务必传 focus」;full模式让它「读全文」。目的是别让 agent 去用一个不存在的工具。 - 策略附言 policy_addendum(
:829-868):按 egress scope 往 prompt 里塞一段——STRICT 只留主web_search、PRIVATE_ONLY 关公网引擎、PUBLIC_ONLY 关本地库。这是「关掉计时侧信道」的另一半:工具清单在_build_tools里已被预过滤,prompt 再明说「别的search_*不存在」,agent 就不会去试探被禁引擎、denial 路径的延迟也就不泄露策略状态(§5 会展开)。 - 流式消费只为「可观测」(
:918-1100):从agent/model/tools三种 chunk 里抽 AIMessage,把「🔍 Searching PubMed for …」「📄 From the web: …」和心跳文案上报给 UI。这些不参与决策,纯粹让用户看见 agent 在想什么、走到第几步、攒了几条源。
4. 线程安全的结果收集器:引用序号不重号的关键
这节讲 SearchResultsCollector——主 agent 和多个并行子 agent 同时往里塞结果,它要保证引用序号全局唯一,并写进整个系统共享的 all_links_of_system。
4.1 它要解决的小问题
引用编号 [1] [2] [3]… 必须全局连续且不撞号。但结果来自多个源头:主 agent 的 web_search、fetch_content,还有 4 个并行子 agent 各自的搜索。如果每个源头自己数数,就会出现两条源都叫 [3]。
4.2 思路:单锁下用「全局偏移」发号
add_results 把「读取当前长度 → 逐条编号 → 追加」整个过程放进一次锁获取里,编号用的是共享 _all_links 的长度(全局偏移)而不是本批的下标(:76-109):
# 示意,非源码 —— 核心是"全局偏移 + 单锁"
def add_results(self, results, engine_name="web"):
with self._lock: # 整段一把锁,序号绝不重复
start_idx = len(self._all_links) # 全局偏移,而非 len(results)
for i, raw in enumerate(results):
r = dict(raw) # 浅拷贝,不污染引擎原始输出
r["index"] = str(start_idx + i + 1) # 1-based 引用号
if "link" not in r and "url" in r:
r["link"] = r["url"] # 归一化 key,引用处理器认 "link"
self._all_links.append(r) # 写进系统共享表
return start_idx
重点看两处:
start_idx = len(self._all_links)——用共享表的长度当偏移。注释点破原因:详细报告里跨小节也不能重号(:92-94)。- 整个 for 循环在一把锁内完成(
:91),所以即便 4 个子 agent 线程同时add_results,也不会两条结果拿到同一个 index(:82-87docstring)。
4.3 共享引用永不重新赋值
_all_links 指向策略的 all_links_of_system,只追加、从不重新赋值(:60-72 docstring 明说)。这保证子 agent 线程看到的始终是同一个列表对象——它们的写入对主 agent 立即可见。
配套的两个方法:
find_by_url(url):URL 若已被收录就返回它已有的 1-based 序号(:111-120)。抓取工具靠它做去重:同一 URL 二次抓取时复用旧引用号,agent 看到的引用稳定。reset():清 per-call 状态,但故意保留_all_links(:122-127)——详细报告每写一个小节analyze_topic都reset一次,可跨小节的全局引用表要留着。
5. 工具集构建:agent 的四类「手脚」
这节讲 _build_tools(:619-776)怎么把工具箱拼出来,以及每类工具的模式。agent 能做什么,完全由这个清单决定。
5.1 四类工具
| 工具 | agent 用它干嘛 | 工厂函数 |
|---|---|---|
web_search | 主引擎通用检索,永远在场 | _make_web_search_tool (:198) |
search_<engine> | 专用引擎(arXiv/PubMed/…),按策略过滤后加入 | _make_specialized_search_tool (:247) |
fetch_content | snippet 不够时抓全文 | build_fetch_tool (fetch/__init__.py:355) |
research_subtopic | 把难题拆成子问题、派并行子 agent | _make_research_subtopic_tool (:291) |
5.2 每个工具都是「每次调用新建引擎」的闭包
web_search 和 search_<engine> 的模式一样:被 @tool 装饰,内部每次调用都新建一个搜索引擎实例,跑完 finally 里 safe_close(:207-240、:257-288)。这样做是为线程安全——子 agent 拿到的是自己的工具实例,不共享可变引擎状态。三步固定:
# 示意,非源码 —— web_search / specialized_search 的共同骨架
@tool
def web_search(query: str) -> str:
engine = create_search_engine(name, llm=model, settings_snapshot=...)
try:
results = engine.run(query) # 跑两阶段检索(见 02)
start = collector.add_results(results, engine_name=name) # 发引用号
return _format_results(results, start) # "[N] 标题 (URL)\n摘要"
except Exception as exc:
return _scrub_tool_error(f"Search error: {exc}") # 洗掉凭据(§6.4)
finally:
safe_close(engine, "...") # 无论成败都释放
专用工具只多两步:装饰后改名成 search_<engine>、把描述换成引擎自己的描述(:286-288)。工具名就是 LLM 在 schema 里看到的名字——这也是为什么下面要在清单层面过滤。
5.3 专用引擎按 egress 策略预过滤(核心安全设计)
这是原始「LangGraph 静默扩权」问题的正解。工厂 PEP 会在实例化时拦下越界引擎,但那是运行时检查——LLM 仍然在 schema 里看得到那个被禁工具名,而且调用被拒的延迟会泄露策略状态(计时侧信道)。所以过滤放在组装工具清单这一步(:657-665 注释):被禁工具压根不进 create_agent,LLM 永远不知道它存在。
过滤链(:666-757),对每个候选引擎逐条判:
候选引擎 name
│ name == 当前主引擎? ── 是 → 跳过(主引擎已单独加)
│ config.agent_enabled == False? ── 是 → 跳过(用户关掉的私有集合)
│ scope == STRICT? ── 是 → 跳过(STRICT 只留主 web_search)
│ 是 retriever? ── 是 → evaluate_retriever(name, ctx)
│ 否 → evaluate_engine(name, ctx, metadata=config)
│ decision.allowed == False? ── 是 → 记 policy_audit 日志 + 跳过
▼ 通过 → 加进工具箱(描述拼上 strengths[:2])
策略上下文由 _build_egress_context() 一次性算好、贯穿所有工具工厂(:619-631)——因为子 agent 线程不继承 thread-local 状态,必须把 ctx 显式塞进闭包,子 agent 才能拿到和主 agent 一样的策略(:578-617)。这里有个正确性细节:主引擎必须用 search.tool 推导(和工厂 PEP 同法),而非引擎类名启发式——否则私有集合主引擎会被错分类、公网引擎漏过滤(:598-607 注释)。
5.4 research_subtopic:把工具变成「派活」
它本身是个工具,但调用它等于再开一批 agent。签名 subtopics: list[str],让 LLM 一次传 2-5 个聚焦子问题(:311-314)。超过 MAX_SUBTOPICS(8)截断(:319-320)。子 agent 的工具箱是精简版:只有 web_search + fetch_content,不能再递归派子 agent(:342-361),避免无限分裂。并行编排细节在下一节。
6. 子 agent 并行:线程池、超时、上下文传播、错误清洗
这节讲 research_subtopic 内部——它是全章工程含量最高的地方:要并发、要传对上下文、要超时兜底、要洗掉凭据。
6.1 并行度与顺序保持
用 stdlib ThreadPoolExecutor 跑,worker 数取 min(MAX_SUBAGENT_WORKERS, 子问题数)(:423-424),即最多 4 并发。虽然并发执行,但结果按原始子问题顺序拼回——用 ordered_results dict 收,最后按 subtopics 原序拼成 ## 子问题\n结果(:422-466)。所以 agent 拿到的多子问题报告顺序是确定的。
6.2 两层超时兜底
超时防线有两层,都用 SUBAGENT_TIMEOUT_SECONDS(1800s):
as_completed(futures, timeout=...)+future.result(timeout=...)——单个子 agent 超时,标「timed out」继续收别的(:430-447)。- 外层再兜一次
except TimeoutError——如果as_completed自己超时(还有 future 没完成),把没结果的子问题统一标超时(:448-458)。
加上 run_subagent 内部对 GraphRecursionError 的捕获(子 agent 撞递归限时返回「达到迭代上限,以上是部分发现」,:387-388),任何一个子 agent 出事都不会拖垮整批。
6.3 egress 上下文与搜索上下文的传播(最容易漏的地方)
ThreadPoolExecutor 的 worker 既不继承 ContextVar,也不继承 threading.local——这带来两个必须手动补的传播:
其一,搜索上下文(带用户 DB 密码)。 在主线程上捕获一次 get_search_context()(:401),每个 worker 进去再用 search_context(...) 重新设上、退出时清掉(:417-420)。不这么做,子 agent 重建搜索引擎、注册用户文档集合时会因拿不到加密 DB 密码而报「Unknown search engine 'collection_…'」。这填的是兄弟策略用 @preserve_research_context 装饰器填的同一个洞(:394-401 注释)。
其二,PEP-578 审计钩子(egress 二道网)。 threading.local 里那个审计钩子在 worker 里是失效的,所以要在 worker 生命周期内用 active_egress_context(egress_context) 重新武装一遍,让子 agent 的出站(LLM、搜索、抓取)在 PRIVATE_ONLY/STRICT 下和主线程有同等的兜底防护(:403-414)。egress_context 为 None 时这是 no-op(fail-open 构建)。
# 示意,非源码 —— worker 包裹层:先武装 egress,再套搜索上下文
def _run_subagent_with_egress(topic):
with active_egress_context(egress_context): # 重装 PEP-578 审计钩子
if captured_search_context is not None:
with search_context(captured_search_context): # 重设 DB 密码
return run_subagent(topic)
return run_subagent(topic)
6.4 错误清洗:凭据不进 LLM、不进用户输出
所有面向 agent/用户的错误串都过 _scrub_tool_error(:50-52),它调 sanitize_error_for_client,先在完整未截断的串上洗凭据,再截到 500 字符(_TOOL_ERROR_MAX_LEN,:47)。为什么是 500 而非 HTTP 客户端默认的 200:这些串既喂 agent 推理、又喂 ErrorReporter 的模式匹配,截太狠会把「可归类的错误信号」切掉(:44-47 注释)。
搜索引擎异常里可能嵌着带 API key 的请求 URL,所以 web_search/specialized_search 的 except 分支必然经过清洗(:231-236);子 agent 失败、超时的消息同理(:389-391、:443-447)。
7. 收尾与合成:从「一堆源」到「带引用的报告」
这节讲 agent 循环结束后怎么落地成结果。三种退出路径都汇到 _finalize。
7.1 三条退出路径
analyze_topic 的 stream 循环有三种收场(:1102-1126):
| 情形 | 处理 | 位置 |
|---|---|---|
| 正常结束,agent 给了 final answer | 直接用 | :1034-1036 |
撞递归限 GraphRecursionError | 若没 final,_synthesize_from_collector 兜底合成 | :1102-1107 |
| 其它异常 | 有结果就兜底合成,没结果才 _error_result | :1108-1114 |
兜底合成 _synthesize_from_collector(:1130-1157):agent 被打断但收集器里已有源时,取前 20 条 [index] 标题: snippet 塞进 prompt,让 model 强行合出一个答案。这保证「跑一半也有东西交」,而不是空手而归。
7.2 _finalize:套引用、组装返回
_finalize(:1159-1240)做三件事:
- 引用处理:有结果时调
citation_handler.analyze_followup(...),把 agent 的原始答案 + 所有搜索结果交给引用处理器,拿回带引用标注的 content 和 documents(:1179-1195)。引用/合成的通用机制见 04。失败则退回 agent 原始答案(:1192-1195)。 - 格式化源:委托基类
_format_citations(:1197-1200)。 - 组装推理轨迹:从
agent_messages里抽每条 assistant 消息的 content 和 tool_calls,拼成reasoning_trace(:1202-1214)——这就是 UI 上「agent 想了什么、调了哪些工具」的可回放记录。
返回一个大 dict:findings / formatted_findings / current_knowledge / sources(去重后的 URL 集)/ reasoning_trace / error(:1222-1240)。这套字段形状和其它策略一致,好让上层 research engine 统一消费(见 01)。
7.3 错误结果的统一形 状
_error_result(:1255-1273)构造一个「空但形状完整」的返回:findings=[]、error=<消息>、其余字段给空值。_format_agent_error(:1242-1253)会给异常加 Agent error: <类型>: 前缀——前缀不含密钥、且在任何截断之前,好让 ErrorReportGenerator 的模式表能按异常类型匹配。
8. 边界与局限
- 依赖模型支持 tool calling。
create_agent对不支持工具调用的模型会直接失败,策略捕获后返回错误结果(:895-908)。 - 推理标签泄露(cosmetic)。 如 §3.1 所述,这条 agent 循环里推理模型的
<think>标签不会被剥掉。已知限制,不崩(:889-894)。 - 子 agent 不能再分裂。 子 agent 只拿到
web_search+fetch_content,没有research_subtopic(:342-361),分解只有一层。 - 合成/引用的细节不在本章。 引用处理器、quick vs detailed 报告见 04;两阶段检索、引擎池与过滤见 02;默认流水线骨架见 01。
9. 代码地图(导航索引)
用符号名 grep 定位,比行号抗上游漂移。
| 主题 | 文件路径 | 符号 |
|---|---|---|
| 策略类 / 主入口 | advanced_search_system/strategies/langgraph_agent_strategy.py | LangGraphAgentStrategy / analyze_topic |
| 迭代常量 | 同上 | DEFAULT_MAX_ITERATIONS / MIN_ITERATIONS / MAX_SUBTOPICS / MAX_SUBAGENT_WORKERS / SUBAGENT_TIMEOUT_SECONDS |
| 装配 | 同上 | LangGraphAgentStrategy.__init__ / _resolve_engine_name |
| 线程安全收集器 | 同上 | SearchResultsCollector / add_results / find_by_url / reset / results |
| 工具箱组装 | 同上 | _build_tools / _build_egress_context |
| 通用检索工具 | 同上 | _make_web_search_tool / _make_specialized_search_tool / _format_results |
| 并行子 agent | 同上 | _make_research_subtopic_tool / run_subagent / _run_subagent_with_egress |
| 收尾合成 | 同上 | _synthesize_from_collector / _finalize / _format_agent_error / _error_result |
| 错误清洗 | 同上 | _scrub_tool_error / _TOOL_ERROR_MAX_LEN |
| 抓取工具 | advanced_search_system/tools/fetch/__init__.py | build_fetch_tool / FETCH_MODES / _make_summary_fetch_tool / _register_in_collector / _enforce_url_policy |
| 工具基类 | advanced_search_system/tools/base_tool.py | BaseTool / TYPE_MAP(安全类型映射,防 eval RCE) |
| 上下文传播 | utilities/thread_context.py | get_search_context / search_context |