引言：什么是 DeepAgents？

DeepAgents 是 LangChain 框架中一个强大的多智能体系统，它允许开发者构建由多个专门化 AI 智能体组成的协作网络，以解决复杂的、多步骤的任务。与传统的单一链式调用不同，DeepAgents 模拟了现实世界中的团队协作，每个智能体可以扮演不同的角色（如研究员、写手、审核员、代码工程师等），并通过内部对话和工具调用，共同完成一个最终目标。

本速通指南将带你快速上手 DeepAgents 的核心概念、搭建你的第一个多智能体系统，并探讨其最佳实践与应用场景。

核心概念

在深入代码之前，理解以下几个关键概念至关重要：

1. 智能体 (Agent): 一个具备特定指令、工具和 LLM 后端的独立执行单元。它是完成任务的基本“员工”。
2. 主管 (Supervisor): 一个特殊的智能体，负责协调工作流。它接收用户请求，将其分解为子任务，并分配给最合适的“员工”智能体执行，最后汇总结果。
3. 团队 (Team): 由一名主管和若干成员智能体组成的协作单元。这是 DeepAgents 的基本组织形式。
4. 工具 (Tools): 智能体可以调用的函数，用于获取信息或执行操作（如网络搜索、代码执行、数据库查询）。
5. 状态 (State): 在团队协作过程中传递的消息和上下文信息。它记录了整个对话和工作流程。

环境准备

首先，确保你已安装必要的库并配置好 API 密钥。

pip install langchain langchain-openai

设置你的 OpenAI API 密钥（或其他兼容后端的密钥）：

import os
from langchain_openai import ChatOpenAI

os.environ["OPENAI_API_KEY"] = "your-api-key-here"

# 初始化一个 LLM，后续供智能体使用
llm = ChatOpenAI(model="gpt-4o", temperature=0)

构建你的第一个多智能体团队

让我们构建一个简单的“内容创作团队”，包含一个“研究员”和一个“写手”。

步骤 1：定义成员智能体

每个智能体需要明确的名称、指令和可用的工具。

from langchain.agents import AgentExecutor, create_openai_tools_agent
from langchain_core.prompts import ChatPromptTemplate, MessagesPlaceholder
from langchain_community.tools import DuckDuckGoSearchRun

# 工具定义：一个简单的网络搜索工具
search_tool = DuckDuckGoSearchRun()

# 1. 研究员智能体
researcher_prompt = ChatPromptTemplate.from_messages([
    ("system", "你是一名专业的研究员。你的职责是根据用户提供的话题，使用搜索工具查找最新、最相关的信息，并整理成简洁的要点。请确保信息的准确性和时效性。"),
    MessagesPlaceholder(variable_name="chat_history"),
    ("human", "{input}"),
    MessagesPlaceholder(variable_name="agent_scratchpad"),
])

researcher_agent = create_openai_tools_agent(llm, [search_tool], researcher_prompt)
researcher_executor = AgentExecutor(agent=researcher_agent, tools=[search_tool], verbose=True)

# 2. 写手智能体
writer_prompt = ChatPromptTemplate.from_messages([
    ("system", "你是一名专业的科技文章写手。你的职责是根据研究员提供的要点，撰写一篇结构清晰、语言流畅、吸引人的博客文章。"),
    MessagesPlaceholder(variable_name="chat_history"),
    ("human", "{input}"),
    MessagesPlaceholder(variable_name="agent_scratchpad"),
])
# 写手可能不需要搜索工具，只使用 LLM 能力
writer_agent = create_openai_tools_agent(llm, [], writer_prompt)
writer_executor = AgentExecutor(agent=writer_agent, tools=[], verbose=True)

步骤 2：定义主管与创建团队

主管负责路由任务。我们需要定义一个主管智能体，并明确团队成员及其职责。

from langchain.agents import AgentExecutor, create_openai_tools_agent
from langchain_core.prompts import ChatPromptTemplate

# 主管智能体
supervisor_prompt = ChatPromptTemplate.from_messages([
    ("system", """你是一名团队主管。你需要根据用户的问题，决定由哪位团队成员来处理，或者是否需要所有成员协作。
团队成员及职责：
- 研究员：擅长搜索和整理信息。当问题需要查找资料、数据或最新动态时，交给他。
- 写手：擅长将信息组织成优美的文章。当需要将要点或资料转化为正式文稿时，交给他。

请只输出团队成员的名字，例如：“研究员”或“写手”。如果任务需要协作，请按顺序输出，如：“研究员，写手”。
现在，请针对以下用户问题做出决定：{input}"""),
])

supervisor_agent = create_openai_tools_agent(llm, [], supervisor_prompt)
supervisor_executor = AgentExecutor(agent=supervisor_agent, tools=[], verbose=True)

# 团队成员映射
team_members = {
    "研究员": researcher_executor,
    "写手": writer_executor,
}

步骤 3：实现协作循环

现在，让我们实现一个简单的主循环，让主管根据用户输入分配任务，成员依次执行。

from langchain_core.messages import HumanMessage, AIMessage
import json

def run_team_workflow(user_input: str):
    """运行团队协作工作流"""
    print(f"\n=== 用户请求 ===")
    print(user_input)

    # 1. 主管决定路由
    route_result = supervisor_executor.invoke({"input": user_input})
    next_agent_name = route_result["output"].strip()
    print(f"\n=== 主管分配 ===")
    print(f"下一个执行者：{next_agent_name}")

    # 2. 根据路由结果，调用相应的成员智能体
    if "," in next_agent_name:
        # 需要多个成员协作
        agent_names = [name.strip() for name in next_agent_name.split(",")]
    else:
        agent_names = [next_agent_name]

    chat_history = [] # 简单的历史记录，可用于传递上下文
    final_result = ""

    for agent_name in agent_names:
        if agent_name in team_members:
            print(f"\n=== {agent_name} 开始工作 ===")
            # 构建成员智能体的输入。这里简单地将用户原始输入和历史传递过去。
            # 更复杂的实现可以将上一个智能体的输出作为下一个的输入。
            member_input = user_input if not final_result else f"基于以下信息进行创作：{final_result}"
            member_result = team_members[agent_name].invoke({
                "input": member_input,
                "chat_history": chat_history
            })
            output = member_result["output"]
            print(f"输出：{output}")
            final_result = output # 这里简化处理，实际可根据需要累积或组合结果
            # 将本次交互加入历史
            chat_history.extend([
                HumanMessage(content=member_input),
                AIMessage(content=output)
            ])
        else:
            print(f"警告：未知的团队成员 '{agent_name}'")

    print(f"\n=== 最终结果 ===")
    return final_result

# 运行示例
if __name__ == "__main__":
    result = run_team_workflow("写一篇关于 LangChain DeepAgents 最新特性的简短博客介绍，要求包含实际应用场景。")
    print(result)

进阶：使用 LangGraph 构建健壮的工作流

上述循环是一个简化示例。对于生产环境，LangChain 官方推荐使用 LangGraph 来构建具有状态管理和复杂循环的多智能体系统。它能更优雅地处理智能体间的对话、条件分支和循环。

from langgraph.graph import StateGraph, END
from typing import TypedDict, Annotated
from langchain_core.messages import AnyMessage, HumanMessage
import operator

# 1. 定义状态结构
class AgentState(TypedDict):
    messages: Annotated[list[AnyMessage], operator.add] # 消息列表，会自动追加
    next: str # 下一个要执行的智能体名称

# 2. 定义每个智能体的节点函数
def researcher_node(state: AgentState):
    """研究员节点"""
    # 从状态中获取最新的人类消息作为输入
    last_message = state["messages"][-1]
    researcher_result = researcher_executor.invoke({"input": last_message.content})
    return {"messages": [AIMessage(content=researcher_result["output"])], "next": "supervisor"}

def writer_node(state: AgentState):
    """写手节点"""
    # 写手可以基于所有历史消息（包括研究员的输出）进行创作
    combined_input = "\n".join([msg.content for msg in state["messages"] if isinstance(msg, HumanMessage)])
    writer_result = writer_executor.invoke({"input": combined_input})
    return {"messages": [AIMessage(content=writer_result["output"])], "next": "supervisor"}

def supervisor_node(state: AgentState):
    """主管路由节点"""
    # 主管根据最新的人类消息决定下一步
    last_human_message = [msg for msg in state["messages"] if isinstance(msg, HumanMessage)][-1]
    route_result = supervisor_executor.invoke({"input": last_human_message.content})
    next_agent = route_result["output"].strip()
    return {"next": next_agent}

# 3. 构建图
workflow = StateGraph(AgentState)

# 添加节点
workflow.add_node("supervisor", supervisor_node)
workflow.add_node("researcher", researcher_node)
workflow.add_node("writer", writer_node)

# 设置入口点
workflow.set_entry_point("supervisor")

# 根据主管的决策，动态路由到成员节点或结束
def route_next(state: AgentState):
    next_agent = state["next"]
    if next_agent == "研究员":
        return "researcher"
    elif next_agent == "写手":
        return "writer"
    elif next_agent == "FINISH":
        return END
    else:
        # 默认返回主管重新决策
        return "supervisor"

workflow.add_conditional_edges(
    "supervisor",
    route_next,
)
# 成员执行完后，都回到主管节点进行下一步协调
workflow.add_edge("researcher", "supervisor")
workflow.add_edge("writer", "supervisor")

# 编译图
app = workflow.compile()

# 4. 运行图
initial_state = AgentState(messages=[HumanMessage(content="写一篇关于AI代理的博客")], next="supervisor")
final_state = app.invoke(initial_state, config={"recursion_limit": 10})
print(final_state["messages"][-1].content)

最佳实践与常见问题

1. 明确智能体职责：给每个智能体清晰、无歧义的 system prompt 和工具集，避免角色重叠。
2. 管理上下文长度：多轮对话容易导致上下文膨胀。考虑使用摘要、只保留关键消息或增加外部记忆存储。
3. 错误处理：智能体调用工具或 LLM 可能失败。在节点函数和图结构中增加重试和降级逻辑。
4. 控制成本与延迟：每个智能体调用都是一次 LLM API 请求。优化工作流，避免不必要的循环，并考虑使用更快的模型处理简单任务。
5. 测试与评估：为你的多智能体系统设计评估标准，如任务完成度、结果质量、步骤数等，并进行充分测试。