使用 LangChain 的 Deep Agents 框架与 Elasticsearch 进行系统性研究

Agent Builder 现已正式发布 (GA)。立即开始 Elastic Cloud 试用，并在此处查看 Agent Builder 的文档。

耗时较长的研究查询，往往需要规划、委派，还要跨多个来源交叉验证。一个带了工具的单一 Agent，处理简单查找还行，但一到上下文限制就捉襟见肘，更别提验证结论了。这篇文章要展示的，就是怎么用 LangChain 的 Deep Agents 框架，搭建一条完整的研究流水线——它能做到这些：

• 用编排器管理的 TODO 列表来规划研究角度
• 把每个研究角度委派给有独立上下文的专业子 Agent
• 把结构化研究结果存到 Elasticsearch 里，支持语义搜索
• 配置一个 Elastic Agent Builder Agent，让团队不用写查询就能探索结果

研究流水线将如何工作？

举个例子，抛出一个复杂问题：“哪种 LLM 最适合小鼠癌细胞的生化分析？” 流水线会生成一套结构化、经过交叉验证的研究结果，存进 Elasticsearch，团队成员随时可以在 Kibana 里探索。

整条流水线分五步走：

1. 规划：编排器读入问题，写一个 TODO 列表，把问题拆成三到四个研究角度（比如性能基准、领域特定能力、成本和可访问性）。
2. 研究：针对每个角度，一个专业的子 Agent 搜遍网络，把每条发现配上结构化元数据（证据类型、相关性评分、来源可信度）一并索引到 Elasticsearch。
3. 评估：评估器子 Agent 查询 Elasticsearch，按角度聚合发现，评估证据质量。
4. 审查：交叉审查器子 Agent 读取所有发现，标记矛盾之处，识别不同来源之间的一致性。
5. 探索：自动配置一个 Elastic Agent Builder Agent，团队用自然语言就能问研究结果相关的问题。

流程图展示了选择 LLM 的五步研究过程，包括规划、研究、评估、审查和探索阶段，每个步骤之间都有箭头连接，最后一个提示询问哪个角度的证据最强。

流程图展示了选择 LLM 的五步研究过程，包括规划、研究、评估、审查和探索阶段，每个步骤之间都有箭头连接，最后一个提示询问哪个角度的证据最强。

先决条件

• Elasticsearch 集群 9.3
• Python 3.10
• Elasticsearch、Google Gemini 和 Ta vily 的 API 密钥
• 集群的 KIBANA_URL

什么是 LangChain Deep Agents？

Deep Agents 是 LangChain 的 Agent harness，一个基于 LangGraph 构建的预组装框架。它给了 Agent 一些纯 LLM + 工具不具备的开箱即用能力：读写文件、用结构化 TODO 列表管理长期任务、生成有独立上下文窗口的专业子 Agent、在多个步骤中持久化状态。

它在 LangChain 生态系统中的位置，可以这样看：

Deep Agents 不是 LangGraph 的替代品，而是在 LangGraph 之上提供预组装架构的更高层级。当你调用 create_deep_agent()，得到的是一张编译好的 LangGraph 图，完整支持流式传输、持久化和检查点。

跟纯 LangGraph Agent 相比，Deep Agents 主要多了这几样：

• write_todos：内置工具，强制编排器在执行前把复杂任务分解。这算是一种上下文工程策略，不只是视觉辅助——它能让长期运行的工作流保持在正轨上。
• task：内置工具，允许编排器生成子 Agent，每个子 Agent 有自己独立的上下文窗口。编排器的上下文保持清爽，专家们则各自深入工作。
• 文件 I/O 工具：read_file、write_file、edit_file 等，把大量结果卸载到磁盘，而不是全塞在上下文里。

对于系统性研究查询，一个带工具的普通 Agent 能搜索并存储发现，但它做不到规划研究角度、把每个角度委派给专家、也不交叉评估结果。Deep Agents 自动实现了这套模式。

设计研究索引

这条流水线里有个关键决策：怎么存储研究发现。纯文本不行——你没法回答“哪些发现有同行评审证据？”或“哪些研究角度的来源最可靠？”这类问题。

解决方案是把结构化元数据字段跟一个由 Jina Embeddings v5 支持的 semantic_text 字段结合起来：

代码语言：ja vascript
复制
INDEX_NAME = "deep-agent-research"
INFERENCE_ID = ".jina-embeddings-v5-text-small"

index_body = {
    "mappings": {
        "properties": {
            "query": {"type": "text", "copy_to": "semantic_field"},
            "source": {"type": "keyword"},
            "title": {
                "type": "text",
                "fields": {"keyword": {"type": "keyword"}},
                "copy_to": "semantic_field",
            },
            "content": {"type": "text", "copy_to": "semantic_field"},
            "timestamp": {"type": "date"},
            "tags": {"type": "keyword"},
            "research_angle": {"type": "keyword"},
            "evidence_type": {"type": "keyword"},
            "relevance_score": {"type": "float"},
            "source_credibility": {"type": "keyword"},
            "semantic_field": {
                "type": "semantic_text",
                "inference_id": INFERENCE_ID,
            },
        }
    }
}

copy_to 模式把 query、title、content 字段路由到同一个 semantic_text 字段。Elasticsearch 在索引时自动生成 embeddings，不需要额外搭建摄入流水线。结构化元数据字段（research_angle、evidence_type、relevance_score、source_credibility）独立保留，用来做过滤和聚合。

这种设计允许你在单个查询里按角度过滤、按可信度分组，同时还能对所有发现做语义搜索。

工具集

流水线里的每个 Agent 只拿到它需要的工具。这样能让子 Agent 的行为可预测，也降低了专家越界干活的风险。

三个工具撑起整条流水线：

• web_search：封装 Ta vily，用来检索和格式化网络结果。只有研究员子 Agent 能用。
• store_finding：把结构化文档索引到 Elasticsearch。它强制执行元数据 schema——调用者必须提供 research_angle、evidence_type（benchmark、case_study、peer_reviewed、expert_opinion）、relevance_score（1.0-10.0）和 source_credibility（peer_reviewed、preprint、industry_report、blog、documentation）。也只有研究员子 Agent 能用。
• query_elasticsearch：接受原始 JSON Elasticsearch 查询体，返回响应。这给了 Agent 完整的查询灵活性：匹配查询、聚合、术语过滤器等等。评估器和审查员子 Agent 用这个来分析已存储的发现，而不是写新的发现。

代码语言：ja vascript
复制
@tool
def query_elasticsearch(query_body: str) -> str:
    """运行 Elasticsearch 查询并返回结果。接受包含有效 Elasticsearch 查询体的 JSON 字符串。
    使用此工具搜索、过滤或聚合已存储的发现。
    示例：
    - 匹配所有: {"query": {"match_all": {}}, "size": 5}
    - 按标签聚合: {"size": 0, "aggs": {"by_tag": {"terms": {"field": "tags"}}}}
    - 按角度过滤: {"query": {"term": {"research_angle": "performance"}}}
    """
    es_client.indices.refresh(index=INDEX_NAME)
    body = json.loads(query_body)
    response = es_client.search(index=INDEX_NAME, body=body)
    return json.dumps(response.body, default=str)

query_elasticsearch 的 docstring 里放了示例查询体。这是故意的——LLM 靠 docstring 理解哪些查询是可行的，具体例子能提升子 Agent 的输出质量。

编排器和子 Agent

流水线有四个阶段：规划、研究、评估、审查。编排器通过委派给三个专业子 Agent 来驱动所有这些阶段。

先定义子 Agent。每个子 Agent 只拿它需要的工具。angle-researcher 不能查询发现；finding-evaluator 和 cross-reviewer 不能存储新发现。这种分离防止评估器在评估时偷偷加发现，把结果搞乱。

代码语言：ja vascript
复制
from deepagents import create_deep_agent
from langchain_google_genai import ChatGoogleGenerativeAI

model = ChatGoogleGenerativeAI(model="gemini-3.1-flash-lite-preview", temperature=0)

angle_researcher = {
    "name": "angle-researcher",
    "description": (
        "研究研究查询的特定角度。搜索网络，评估来源，并将每个发现 "
        "与结构化元数据一起存储在 Elasticsearch 中。"
    ),
    "system_prompt": (
        "你研究科学查询的 ONE 个特定角度。"
        "1. 使用 web_search 查找给定角度的 3-5 个相关来源。"
        "2. 对于 EACH 有用的来源，调用 store_finding 并提供："
        " - research_angle: 分配给你的角度"
        " - evidence_type: 'benchmark', 'case_study', 'peer_reviewed', 或 'expert_opinion'"
        " - relevance_score: 1.0-10.0，基于它直接解决查询的程度"
        " - source_credibility: 'peer_reviewed', 'preprint', 'industry_report', 'blog', 或 'documentation'"
        "3. 返回你针对此角度发现的简要摘要。"
    ),
    "tools": [web_search, store_finding],
}

finding_evaluator = {
    "name": "finding-evaluator",
    "description": (
        "评估和分析已存储在 Elasticsearch 中的发现。按研究角度聚合，"
        "计算平均分数，并识别哪些角度具有最强的证据。"
    ),
    "system_prompt": (
        "你分析存储在 Elasticsearch 中的研究发现。"
        "使用 query_elasticsearch 来："
        "1. 按 research_angle 聚合发现并计算平均 relevance_score。"
        "2. 按 source_credibility 聚合以评估证据质量。"
        "3. 按 evidence_type 聚合以了解证据的组合。"
        "4. 识别哪些角度具有最多和最强的发现。"
        "返回结构化的评估摘要。"
    ),
    "tools": [query_elasticsearch],
}

cross_reviewer = {
    "name": "cross-reviewer",
    "description": (
        "审查所有已存储的发现，以识别研究角度之间的矛盾、共识和差距。"
        "生成最终评估。"
    ),
    "system_prompt": (
        "你是一个研究发现的批判性审查员。"
        "使用 query_elasticsearch 检索所有发现，然后："
        "1. 识别出现在多个角度的声明（共识）。"
        "2. 标记来源之间的任何矛盾。"
        "3. 注意哪些角度缺乏同行评审证据。"
        "4. 根据证据生成最终的排名推荐。"
        "保持怀疑态度。优先考虑同行评审的来源，而不是博客。"
    ),
    "tools": [query_elasticsearch],
}

接着创建编排器，跟子 Agent 连上。编排器的系统提示强制执行四阶段序列，用编号程序写清楚——这是从长期运行 Agent 拿到可靠行为的关键。子 Agent 有独立的上下文窗口：每次调用 task 都启动一个全新上下文，编排器的上下文不会随着每次委派而膨胀。这正是 Deep Agents 在长期运行工作方面的核心架构优势。

代码语言：ja vascript
复制
agent = create_deep_agent(
    model=model,
    tools=[web_search, store_finding, query_elasticsearch],
    subagents=[angle_researcher, finding_evaluator, cross_reviewer],
    system_prompt=(
        "你是一个系统性研究编排器。对于每个查询："
        "1. 规划 (PLAN): 使用 write_todos 将查询分解为 3-4 个研究角度 "
        "（例如，'性能基准'、'成本和可访问性'、'领域特定能力'）。"
        "2. 研究 (RESEARCH): 对于每个角度，使用 task 工具委派给 'angle-researcher' 子 Agent。 "
        "每次子 Agent 调用都应指定角度和原始查询。"
        "3. 评估 (EVALUATE): 委派给 'finding-evaluator' 以分析已存储的发现 "
        "并评估跨角度的证据质量。"
        "4. 审查 (REVIEW): 委派给 'cross-reviewer' 以交叉检查发现，标记 "
        "矛盾之处，并生成共识分数。"
        "5. 综合 (SYNTHESIZE): 撰写最终报告，根据证据总结最佳答案，"
        "并注明置信水平和差距。"
        "完成每个 TODO 后将其标记为已完成。"
    ),
)

运行流水线

流水线以流式传输方式运行，这样能实时看到进度。加上 subgraphs=True 标志，来自子 Agent 的事件也会暴露出来，跟编排器自身的事件混在一起：

代码语言：ja vascript
复制
research_query = ("What is the best LLM for biochemical analysis of cancer cells in mice?")

for chunk in agent.stream(
    {"messages": [{"role": "user", "content": research_query}]},
    stream_mode="updates",
    subgraphs=True,
    version="v2",
):
    if chunk["type"] == "updates":
        is_subagent = any(segment.startswith("tools:") for segment in chunk["ns"])
        if is_subagent:
            tool_call_id = next(s.split(":")[1] for s in chunk["ns"] if s.startswith("tools:"))
            print(f"Subagent {tool_call_id}: {chunk['data']}")
        else:
            print(f"Main agent: {chunk['data']}")

执行期间，你能看到编排器在委派之前调用 write_todos 并列出研究角度。每次 task 调用都会把任务交给子 Agent，chunk["ns"] 里的事件让你能区分是子 Agent 的输出还是编排器自己的步骤。

代码语言：ja vascript
复制
Main agent: {'PatchToolCallsMiddleware.before_agent': {'messages': Overwrite(value=[HumanMessage(content='What is the best LLM for biochemical analysis of cancer cells in mice?', additional_kwargs={}, response_metadata={}, id='906965ba-4c4c-48cb-8663-6fcfa7b9a2b6')])}}
Main agent: {'model': {'messages': [AIMessage(content=[], additional_kwargs={'function_call': {'name': 'write_todos', 'arguments': '{"todos": [{"content": "Research LLMs specialized in bioinformatics and biomedical data analysis.", "status": "in_progress"}, {"status": "pending", "content": "Investigate LLM capabilities for processing biochemical assay data (e.g., proteomics, transcriptomics) in cancer research."}, {"status": "pending", "content": "Evaluate LLM performance in mouse model studies and preclinical cancer research."}, {"status": "pending", "content": "Synthesize findings to identify the most suitable LLM or approach for biochemical analysis of cancer cells in mice."}]}'}, '__gemini_function_call_thought_signatures__': {'6bba74f0-41b3-4fc3-b512-27529e99d981': 'EjQKMgG Pvb7HjHZXrWuOEjj94E8IYS aAKgZKjvzB8MqxveU5GAhsoq5icieLF6V70r4SfS'}}, ...

这条流水线可以在同一会话里处理多个查询。所有查询的发现都累积在同一个 Elasticsearch 索引中，这让最后的 Agent Builder 探索变得更有意思。比如同时跑“哪种 LLM 最适合生化分析？”和“哪种前端框架最适合 Deep Agent 服务？”，不同领域的发现就存在同一个索引里，Agent Builder 能对比两者的模式。

使用 Elastic Agent Builder 探索发现

流水线跑完后，所有发现都存在 Elasticsearch 里了，但大多数团队成员不会写 Elasticsearch 查询。Elastic Agent Builder 在这里特别合适——它能快速把索引数据转成可用的 Agent，让团队直接用。具体来说，只需要几个 API 调用，就能创建一个基于研究发现的 Agent。我们通过编程方式来创建，这样大家就能在 Kibana UI 里探索这些发现了。

首先创建一个 Elasticsearch Query Language (ES|QL) 工具，用来聚合研究索引：

代码语言：ja vascript
复制
tool_body = {
    "id": "research-findings-tool",
    "type": "esql",
    "description": (
        "查询 deep-agent-research 索引以探索已存储的发现。 "
        "返回按研究角度和来源可信度分组的平均相关性分数和计数。"
    ),
    "configuration": {
        "query": (
            "FROM deep-agent-research "
            "| STATS a vg_score = A VG(relevance_score), count = COUNT(*) BY research_angle, source_credibility"
        ),
        "params": {},
    },
}
resp = requests.post(f"{KIBANA_URL}/api/agent_builder/tools", headers=headers, json=tool_body)

然后创建 Agent Builder Agent，把工具挂上去：

代码语言：ja vascript
复制
agent_body = {
    "id": "research-explorer",
    "name": "Research Explorer",
    "description": "探索和分析来自 deep-agent-research 索引中存储的研究发现。",
    "configuration": {
        "instructions": (
            "你帮助用户探索存储在 Elasticsearch 中的研究发现。 "
            "使用 research-findings-tool 按角度查询发现，"
            "总结证据质量，并回答有关已存储研究的问题。"
        ),
        "tools": [{"tool_ids": ["research-findings-tool"]}],
    },
}
resp = requests.post(f"{KIBANA_URL}/api/agent_builder/agents", headers=headers, json=agent_body)

创建完成后，Research Explorer 就会出现在 Kibana 的 Agent Builder UI 里。团队成员打开它，用自然语言提问就行：

界面展示了一个关于没有同行评审来源的研究角度的查询，包括推理部分、ESQL 代码块、结果计数以及一个总结，指明一个角度只有行业报告而没有同行评审来源。

界面展示了一个关于没有同行评审来源的研究角度的查询，包括推理部分、ESQL 代码块、结果计数以及一个总结，指明一个角度只有行业报告而没有同行评审来源。

这个 Agent 使用 ES|QL 工具查询索引并总结结果，完全不用手动写查询。

这就形成了一个完整的闭环：Deep Agents 做系统性研究，Elasticsearch 存结构化发现，Agent Builder 让团队里的任何人都能访问这些发现。

结论

我们聊了这些：

• Deep Agents 用于长期研究：write_todos 强制先规划再执行，task 工具把每个研究角度委派给有独立上下文窗口的子 Agent。这种模式能处理那种单一 Agent 扛不住的复杂查询。
• Elasticsearch 作为结构化知识存储：把 research_angle、evidence_type、relevance_score、source_credibility 跟 semantic_text 一起存，既能做语义搜索，又能做研究质量的结构化聚合。纯文本存储的话，评估器和审查员 Agent 根本跑不起来。
• Agent Builder 作为访问层：流水线结束时，通过编程方式配置一个 Kibana Agent，发现结果就能直接在 UI 里探索，完全不用手动设置。

这套模式适合任何需要从多个角度研究问题并交叉验证结果的领域：竞争情报、科学文献综述、法规遵从研究、多源事实核查——都适用。

后续步骤

• 在 GitHub 上试试完整笔记本。
• 用 LangGraph 的中断模式给流水线加上人工审批门 (human-in-the-loop)。
• 看看怎么用 Elastic Agent Builder 和 Model Context Protocol (MCP) 构建 Agent 应用的参考架构。
• 扩展 Agent Builder 里的 ES|QL 工具，支持参数化查询，按角度或日期范围过滤。