跳到主要内容

数据截至 (上游 commit c80d325ec761)

研究引擎主线与策略骨架(默认 source-based)

30 秒导读: Local Deep Research(LDR)把"一个问题 → 一份带引用的报告"这件事,拆成一条统一契约:每种研究"策略"都实现同一个方法 analyze_topic(query)。本章讲清这条主线——协调器 AdvancedSearchSystem 如何按名字选出策略、给它套上安全网、再把问题交给它;并逐行精读默认策略 source-based 的完整循环。搜索引擎内部机制在 02,智能体策略在 03,引用与报告在 04。

1. 这节讲什么(一句话主线)​

研究引擎 = 一个协调器 + 一组可互换的策略;所有策略共享同一个契约 analyze_topic(query) -> dict。

把它想成餐厅:

  • 协调器(AdvancedSearchSystem)是领班——不亲自做菜,只负责按你点的名字(strategy_name)叫来对应的厨师,把食材(LLM、搜索引擎、设置快照)交给他,并在上菜前做食品安全检查(egress 安全网)。
  • 策略(source-based / langgraph-agent / …)是厨师——每个厨师做法不同,但交付接口一样:你给一句问题,他还你一份 dict(findings、合成正文、引用清单)。

因为契约统一,换策略对上层几乎零成本——这就是本章要让你记住的那句话:"策略 = 一套 analyze_topic 流水线"。

2. 顶层全景(它大概怎么转)​

一次研究从"一句问题"到"带引用的合成结果",走这条链:

调用方(Web worker / CLI / 程序化 API)
   │  new AdvancedSearchSystem(llm, search, strategy_name="source-based", ...)
   ▼
┌─────────────────────────────────────────────────────────┐
│ AdvancedSearchSystem  (协调器 / 领班)                     │
│   __init__ :                                            │
│     · ensure_snapshot_username   给快照塞 _username      │
│     · create_strategy(...)       工厂按名字造厨师         │
│   analyze_topic(query) :                                │
│     · _arm_egress_backstop       上菜前武装安全网         │
│     · _perform_search  ──► strategy.analyze_topic(query) │
└───────────────────────────────┬─────────────────────────┘
                                 │ 统一契约
                                 ▼
┌─────────────────────────────────────────────────────────┐
│ SourceBasedSearchStrategy  (默认厨师)                     │
│   迭代循环:  生成问题 → 并行搜索 → 累积结果               │
│   收尾:      跨引擎过滤 → 引用偏移合成                     │
│   共享真源:  all_links_of_system(引用的唯一权威清单)     │
└─────────────────────────────────────────────────────────┘

各部件一句话职责:

部件干什么在哪个文件
AdvancedSearchSystem协调器:选策略、注入快照、武装安全网、转调策略search_system.py:69
create_strategy工厂:按 strategy_name(含别名/降级)造出策略实例search_system_factory.py:30
AVAILABLE_STRATEGIESUI 可见的策略清单(单一真源)constants.py:123
BaseSearchStrategy策略抽象基类:进度回调、终止检查、错误响应、引用格式化advanced_search_system/strategies/base_strategy.py:14
SourceBasedSearchStrategy默认策略:迭代搜索 + 跨引擎过滤 + 合成advanced_search_system/strategies/source_based_strategy.py:22
run_parallel_searches把一批问题并发跑,每个 worker 拿到独立 Flask 上下文advanced_search_system/parallel_search.py:45
StandardQuestionGenerator让 LLM 从当前知识产出下一批搜索问题advanced_search_system/questions/standard_question.py:14
FindingsRepository暂存文档/问题,把 findings 格式化成报告文本advanced_search_system/findings/repository.py:30

3. 协调器 AdvancedSearchSystem:它只做三件事​

协调器不含任何检索逻辑。它的价值在于把"选策略、准备环境、兜底安全"这些横切关注点收在一处。

3.1 构造期:选策略 + 修快照​

默认策略就是 source-based(strategy_name: str = "source-based",search_system.py:78)。构造函数做两件关键事:

① 给设置快照补 _username。 settings_snapshot 是一份冻结的设置字典(整个运行期不再变,细节见 05)。用户身份原本存在 system.user 下,但某些快照驱动的消费者只认顶层 _username 键——尤其是 langgraph-agent 每次工具调用都会用这份快照重新造一个搜索引擎,注册用户的文档集合时要用户名。于是构造期用一个纯函数补上:

# search_system.py:21  ensure_snapshot_username(settings_snapshot, username)
# 无 username / 非 dict / 已有 _username → 原样返回(绝不覆盖显式值)
# 否则返回一份浅拷贝,加上 _username(不改原对象)
if username and isinstance(settings_snapshot, dict) and not settings_snapshot.get("_username"):
    return {**settings_snapshot, "_username": username}
return settings_snapshot

这段的用意:在"跑策略"这条最窄的公共入口修一次,Web 运行和程序化 API 就一起修好了(search_system.py:146)。

② 经工厂造策略。 大多数策略走 create_strategy(...)(search_system.py:241);只有 follow-up 系列因为要包一个"委托策略",走单独分支(search_system.py:204-237)。构造好后把 all_links_of_system(空列表)和一堆策略专属参数一并注入——注意 all_links_of_system 是按引用传进去的同一个列表对象,这是后面引用合成的关键(见 §5.5)。

3.2 运行期:analyze_topic 只是给策略套了层安全网​

协调器自己的 analyze_topic 很薄——它不做检索,只在调策略前后加一道 egress(出站)安全网,真正干活的是 _perform_search:

analyze_topic(query, ...)                       # search_system.py:321
  │
  ├─ 生成 search_id(缺省则 uuid4)
  ├─ _armed = _arm_egress_backstop()            # :352  武装安全网(若无人先武装)
  │      try:
  ├─────────  return _perform_search(...)        # :355  真正干活
  │      finally:
  └─ if _armed: clear_active_context()          # :363  只清自己武装的那层

egress 安全网是什么、为什么在这。 LDR 有一套出站管制:研究过程会往外发很多请求(搜索引擎、抓网页),安全层要拦住不该出的流量(完整机制见 06)。主防线是代码里显式的策略执行点(PEP);这里的 _arm_egress_backstop 是第二道防线——基于 PEP-578 审计钩子的兜底:

  • Web worker 在调协调器之前已经武装好了,get_active_context() 非空 → 这里直接返回 False,不重复武装(search_system.py:387)。
  • 但 CLI、新闻调度器、程序化 API 是直接 new 出协调器的,没人先武装 → 这里补上一层,否则整条管线就在没有次级网的情况下裸奔(search_system.py:350)。
  • _arm_egress_backstop 绝不抛异常(兜底失败不能拖垮研究运行);策略不可评估时静默不武装(search_system.py:401-409)。
  • 谁武装谁负责清:只有 _armed == True(是本次调用武装的)才在 finally 里 clear_active_context(),免得复用的线程把上下文泄漏给后续无关任务(search_system.py:358-365)。

3.3 _perform_search:转调策略并回收结果​

这才是把问题交给策略的地方,核心就一行:

# search_system.py:476
result = self.strategy.analyze_topic(query)

前后是进度播报和结果回收:先播报 LLM/搜索引擎信息(search_system.py:452-473),调完策略后:

  • 把策略的 questions_by_iteration 复制回协调器(向后兼容,:480);
  • 只有当两个 all_links_of_system 不是同一对象时才 extend,避免详细报告模式下引用清单翻倍(修复 issue #301,search_system.py:488-491);
  • 把 search_system 自身和 all_links_of_system 塞进 result,供报告生成器回来取引用(:478-479);
  • 最后触发新闻回调(:487-497)。

4. 策略工厂与可用清单​

4.1 工厂 create_strategy:名字 → 实例​

工厂是一长串 if strategy_name_lower in [...] 的分派(search_system_factory.py:30)。每个分支负责一族别名,并从 settings_snapshot 读该策略的专属默认值。三个要点:

① 别名归一。 名字先 .lower(),每族接受多个写法,例如 source-based 同时认 "source-based"、"source_based"、"source_based_search"(:60-64)。

② 已移除策略的降级(inferred 补注:降级目标)。 mcp 和 agentic 两个旧策略在 #4548 被删,但工厂保留它们作为弃用别名,路由到最接近的后继 langgraph-agent,并打一条 warning——这样存量的已保存设置、排队任务、API 调用不会掉进"未知策略"的默认分支:

# search_system_factory.py:342
if strategy_name_lower in ["langgraph-agent", "langgraph_agent", "mcp", "agentic"]:
    if strategy_name_lower in ("mcp", "agentic"):
        logger.warning(f"Strategy {strategy_name!r} was removed (#4548); using 'langgraph-agent' instead.")
    ...  # 造 LangGraphAgentStrategy

③ 未知名兜底 source-based。 任何认不出的名字,记一条 warning,降级为默认 source-based(search_system_factory.py:386-402)。所以引擎"永远能跑",最坏也是退回默认厨师。

4.2 UI 清单 AVAILABLE_STRATEGIES​

工厂认的名字(含别名、内部策略如 news_aggregation)远多于 UI 展示的。给用户看的清单是 constants.py:123 的 AVAILABLE_STRATEGIES,get_available_strategies() 返回它的拷贝(constants.py:261)。当前 UI 面向用户的五种:

name定位
source-based默认;小上下文窗口(<16k)也能跑的综合研究,带内联引用
focused-iteration快速精准问答,输出精简
focused-iteration-standard长文详细输出,需 >16k 上下文
topic-organization把来源按主题聚类
langgraph-agent自主智能体研究(见 03)

注意区分两个清单: AVAILABLE_STRATEGIES 只管 UI 下拉;工厂 create_strategy 才是运行时真正能造出来的全集(含别名和内部策略)。改一处不自动改另一处。

5. 策略抽象基类:所有厨师的公共台面​

BaseSearchStrategy(base_strategy.py:14)是抽象基类,定义了那条契约,并提供每个策略都用得上的公共设施。它只强制一个抽象方法:

# base_strategy.py:119
@abstractmethod
def analyze_topic(self, query: str) -> dict[str, Any]:
    # 返回 dict 至少含:findings / iterations / questions_by_iteration
    #                  / formatted_findings / current_knowledge / (可选 error)
    ...

其余都是共享工具,子类直接用:

5.1 进度回调 _update_progress / set_progress_callback​

策略通过 progress_callback(message, percent, metadata) 往上报进度(base_strategy.py:109)。协调器在构造末尾把自己的回调透传给策略(search_system.py:278),于是 UI 能实时看到"生成问题→并行搜索→过滤→合成"各阶段。基类还封了几个语义化的播报器,如 _emit_question_generation_progress(:201)、_emit_searching_progress(:290)。

5.2 终止检查 check_termination​

用户点"停止"怎么中断一个跑到一半的研究?答案很巧:复用 progress_callback,而不是另开一条终止通道(base_strategy.py:86)。

# base_strategy.py:101
if self.progress_callback:
    self.progress_callback("Checking termination status", None, {"phase": "termination_check"})

research_service.py 里的回调识别 "termination_check" 这个 phase,检查停止标志后立即返回(不写日志、不发 socket)。这样省掉了在 ~15 处接线点再传一个专用回调的维护成本。策略在每轮迭代开头调 self.check_termination()(source-based 在 :219)。

5.3 标准错误响应 _create_error_response​

出错时返回一个形状统一的 dict(base_strategy.py:244),同时带 questions 和 questions_by_iteration 两个键——因为历史上两类消费者读的键不同,都留着免得任一方崩。

5.4 引用格式化 _format_citations​

给正文追加一段 Markdown 的 ## Sources 书目(base_strategy.py:262)。注意:这是给 mcp / langgraph-agent 用的通用工具;默认 source-based 不走这条,它有自己更精细的引用偏移逻辑(见 §5.5)。抽不出链接时原样返回正文。

6. 精读默认策略 source-based 的完整循环​

这是本章的重点。SourceBasedSearchStrategy.analyze_topic(source_based_strategy.py:159)把"问题 → 带引用合成"演成一条流水线。先看骨架,再看每一步。

analyze_topic(query):
  ① 起手:记下引用偏移 total_citation_count_before_this_search = len(all_links_of_system)
  ② 迭代循环  for iteration in 1..N:              # N = get_setting("search.iterations")
        check_termination()
        ├─ 生成问题(iteration==1 与 >1 逻辑不同)
        ├─ run_parallel_searches(问题们)           # 并行检索
        └─ 结果 extend 进 accumulated_...(本地累积)
  ③ 跨引擎过滤(可关):accumulated → final_filtered_results
  ④ all_links_of_system.extend(final_filtered_results)   # 写入共享真源
  ⑤ 合成:citation_handler.analyze_followup(..., nr_of_links=偏移)
  ⑥ FindingsRepository 收文档 + 格式化成报告文本
  return dict(findings, formatted_findings, current_knowledge, all_links_of_system, ...)

6.1 迭代次数从哪来(一个小陷阱)​

策略在运行时从快照现读迭代次数,而不是用构造期传进来的 max_iterations:

# source_based_strategy.py:202
iterations_to_run = self.get_setting("search.iterations", 2)

诚实标注: 类 docstring 说"default 5 iterations"(:28),但代码里 get_setting 的兜底是 2。真正生效的是快照里 search.iterations 的值;两者不一致时以代码为准。每轮问题数实际读的键是 search.questions_per_iteration(iteration 1 兜底 5、后续兜底 2,:267 / :332),:204 那个 search.questions 变量并不驱动生成。

6.2 第 1 轮:原始 query 逐字 + LLM 问题​

第一轮特殊:除了 LLM 生成的问题,还把用户原话原封不动作为一条搜索(当 search_original_query=True)。

为什么? 因为 LLM 可能把有歧义的 query 理解歪——注释举了个真例:"local deep research" 被 LLM 联想成"Austin 热岛效应"。逐字搜原话,保证至少有一条搜索命中用户真正想问的东西(source_based_strategy.py:237-241)。

# source_based_strategy.py:275  第 1 轮拼问题清单
all_questions = (
    [query] + questions
    if self.search_original_query and query not in questions
    else questions
)

还有个保护:原话太长(超 app.max_user_query_length,默认 300 字符)就跳过逐字搜、只用 LLM 问题,搜完再把开关恢复原状(:249-301)。

6.3 第 2+ 轮:拿累积结果当上下文生成追问​

后续轮次让 LLM 基于已有搜索结果产出下一批问题——这就是"迭代深化"。喂给 LLM 的上下文是累积结果的最近 N 条(search.question_context_limit,默认 30):

# source_based_strategy.py:310
question_context_limit = int(self.get_setting("search.question_context_limit", 30))
source_context = self._format_search_results_as_context(
    accumulated_search_results_across_all_iterations[-question_context_limit:]
)

两个设计细节:

  • 用的是累积结果而非"仅上一轮"——万一某轮被限流返回 0 条,问题生成器仍有上下文可用(:304-306)。
  • _format_search_results_as_context 是纯只读格式化器(:136):只 .get() 读 title/snippet/link 拼字符串,绝不改动输入列表,更不碰共享的 all_links_of_system。

6.4 并行搜索:每个问题一条并发查询​

一轮里的所有问题并发跑。策略把"搜一个问题"包成 search_question,交给 run_parallel_searches:

# source_based_strategy.py:379
completed = run_parallel_searches(
    all_questions,
    search_question,              # 每个问题 → self.search.run(q, research_context=...)
    context_factory=thread_context,   # 每个 worker 拿到独立的 Flask app context
)

run_parallel_searches(parallel_search.py:45)用 ThreadPoolExecutor 跑并发,有个非做不可的坑它替你填了:

Flask 上下文不能跨线程共享。 context_factory(这里是 thread_context)在调用线程上、每个问题调一次,给每个 worker 造一份全新的 app context。若把同一个 context 实例塞给多个并发 worker,Flask 的 token 栈会炸 ValueError: <Token> was created in a different Context(parallel_search.py:16-26 / 91-105)。

返回的是 (query, payload) 列表(完成顺序);策略把各 payload 的结果既做成"问题→结果"字典,又 extend 进本轮列表,最后并入跨轮累积列表(source_based_strategy.py:384-398)。

6.5 跨引擎过滤 + all_links_of_system 这个共享真源​

所有轮跑完后,若开了跨引擎过滤(默认开),用 LLM 对本次调用累积的结果做相关性过滤/重排/重新编号(source_based_strategy.py:415-432,机制细节在 02)。关了则原样保留全部累积结果(:442-447)。

然后把过滤结果写进共享真源 all_links_of_system:

# source_based_strategy.py:452
self.all_links_of_system.extend(final_filtered_results)

为什么这个列表这么重要? 它是整份报告引用的唯一权威来源,而且是跟协调器同一个列表对象(构造期按引用传入)。这带来一个关键能力——详细报告模式下的引用连续编号:

详细报告 = IntegratedReportGenerator 对每个小节各调一次 analyze_topic()

第 1 次调用: 起手记 offset=0   → 写入 [1]..[15],  all_links 现有 15 条
第 2 次调用: 起手记 offset=15  → 写入 [16]..[28], all_links 现有 28 条
             (上一节的结果永不被重过滤、永不被动)

offset 就是这次搜索开始前共享列表的长度(total_citation_count_before_this_search = len(self.all_links_of_system),:178 / :214)。它有两处用途:

  • 传给跨引擎过滤的 start_index,让新结果拿到接续的引用序号(:430);
  • 传给合成器的 nr_of_links,让正文里的 [n] 从上一节续下去(见 §6.6)。

6.6 合成:偏移引用把结果变成带 [n] 的正文​

拿到 final_filtered_results 后,交给引用处理器合成正文:

# source_based_strategy.py:477
final_citation_result = self.citation_handler.analyze_followup(
    query,
    final_filtered_results,
    previous_knowledge="",
    nr_of_links=total_citation_count_before_this_search,   # 引用偏移
)

analyze_followup(question, search_results, previous_knowledge, nr_of_links)(citation_handler.py:98)委托给具体处理器,产出带 IEEE 风格内联引用 [1] [2] 的正文;nr_of_links 保证这些编号从上一节续下去(引用/合成的完整细节在 04)。

一个诚实的边界:没有任何来源时不硬合成。若 final_filtered_results 为空,引用处理器拒绝调 LLM(否则会瞎编引用),直接回一句显式的"没找到来源"(source_based_strategy.py:466-475)。

合成还有个隐藏的副作用利用:analyze_followup 内部会给每个结果 dict 补 "index" 键,而这些 dict 就是 all_links_of_system 里的同一批对象,所以序号自动传导到共享清单(:483-486)。

6.7 收尾:FindingsRepository 攒料并格式化​

最后把合成正文和文档交给 FindingsRepository(findings/repository.py:30):

  • add_documents(documents) 收下 LangChain 文档(:103);
  • set_questions_by_iteration(...) 存下各轮问题,好让报告里能显示(:112);
  • format_findings_to_text(findings, synthesized_content) 调 format_findings 工具把一切拼成最终报告文本(:125);失败则回退成只返回原始合成正文(:161-164)。

analyze_topic 最终返回的 dict(source_based_strategy.py:538)里,current_knowledge 在快速摘要模式下就是最终输出;详细报告模式下 IntegratedReportGenerator 会借 search_system 引用反复回调 analyze_topic,一节一节垒在 all_links_of_system 上。

7. 把三个配角摆正:问题生成器 / 并行器 / findings 仓储​

这三者是 source-based 循环里被 analyze_topic 编排的"配角",各司一职:

配角契约方法在循环里的角色引用
StandardQuestionGeneratorgenerate_questions(current_knowledge, query, n, questions_by_iteration)每轮产出下一批搜索问题;首轮无历史用短提示,有历史用"反思当前知识找缺口"的长提示questions/standard_question.py:17
run_parallel_searches(queries, search_fn, context_factory)把一轮的问题并发跑,替每个 worker 造独立 Flask 上下文parallel_search.py:45
FindingsRepositoryadd_documents / set_questions_by_iteration / format_findings_to_text攒文档与问题,合成后格式化成报告findings/repository.py:30

问题生成器的解析很朴素:让 LLM 按 Q: ... 每行一条输出,再挑出以 Q: 开头的行、截到 questions_per_iteration 条(standard_question.py:46-50)。契约稳定、实现可替换——use_atomic_facts=True 时换成 AtomicFactQuestionGenerator(source_based_strategy.py:130)。

8. 边界与本章不讲的部分​

  • 两阶段检索的引擎内部(self.search.run 里怎么先拿摘要再抓正文、引擎池怎么选、跨引擎过滤器怎么打分)→ 02。本章只把 search.run 当黑盒。
  • langgraph-agent 策略(那条 ~95% SimpleQA 的自主智能体路线)→ 03。本章只讲它在工厂里的别名与降级。
  • 引用处理器与报告生成(analyze_followup 内部、quick vs detailed、Sources 段怎么出)→ 04。
  • 设置快照、加密库、LLM 提供方、Web 编排(settings_snapshot 从哪来、get_setting 底层)→ 05。
  • egress 出站管制的完整策略层(PEP、context_from_snapshot)→ 06。本章只讲协调器如何武装/清除那道兜底网。

一个诚实的坑再强调一次: source-based 的 docstring 与代码在"默认迭代数(5 vs 2)"上不一致,真正生效以 get_setting("search.iterations", 2) 为准。

9. 代码地图(导航索引)​

主题文件路径符号名
协调器主类src/local_deep_research/search_system.pyAdvancedSearchSystem
快照补用户名src/local_deep_research/search_system.pyensure_snapshot_username
武装 egress 兜底网src/local_deep_research/search_system.pyAdvancedSearchSystem._arm_egress_backstop
协调器 analyze 入口src/local_deep_research/search_system.pyAdvancedSearchSystem.analyze_topic
转调策略并回收src/local_deep_research/search_system.pyAdvancedSearchSystem._perform_search
策略工厂src/local_deep_research/search_system_factory.pycreate_strategy
UI 策略清单src/local_deep_research/constants.pyAVAILABLE_STRATEGIES / get_available_strategies
默认搜索工具常量src/local_deep_research/constants.pyDEFAULT_SEARCH_TOOL
策略抽象基类src/local_deep_research/advanced_search_system/strategies/base_strategy.pyBaseSearchStrategy
终止检查src/local_deep_research/advanced_search_system/strategies/base_strategy.pyBaseSearchStrategy.check_termination
标准错误响应src/local_deep_research/advanced_search_system/strategies/base_strategy.pyBaseSearchStrategy._create_error_response
默认策略主循环src/local_deep_research/advanced_search_system/strategies/source_based_strategy.pySourceBasedSearchStrategy.analyze_topic
结果→上下文格式化src/local_deep_research/advanced_search_system/strategies/source_based_strategy.pySourceBasedSearchStrategy._format_search_results_as_context
并行搜索src/local_deep_research/advanced_search_system/parallel_search.pyrun_parallel_searches
问题生成器src/local_deep_research/advanced_search_system/questions/standard_question.pyStandardQuestionGenerator.generate_questions
findings 仓储src/local_deep_research/advanced_search_system/findings/repository.pyFindingsRepository
引用合成入口src/local_deep_research/citation_handler.pyCitationHandler.analyze_followup