手把手搭建DeepAgents项目：从基础到进阶-吉游戏

作者 | 崔皓

审校 | 重楼

在AI开发场景中，很多开发者都有过这样的困扰：普通智能体只能处理简单的单步指令，面对多步骤、长周期的复杂任务时，要么杂乱无章，要么记忆过载，很难达到实际应用需求。而DeepAgents的出现，让智能体从“能执行”升级为“能组织”，成为具备统筹能力的“智能管理者”。更重要的是，借助LangChain和LangGraph工具，我们可以循序渐进地搭建一个属于自己的DeepAgents项目，从基础工具调用到进阶的子智能体协同，每一步都清晰可落地。

先搞懂：普通智能体的“痛点”，就是DeepAgents的“突破口”

在动手搭建项目前，我们先明确核心需求——为什么需要DeepAgents？普通智能体的三大痛点，正是我们搭建项目时需要重点解决的问题，也是DeepAgents的核心价值所在。

日常开发中，我们接触的普通智能体，处理“查天气”“转文本”这类单步任务时毫无压力，但如果让它完成“调研竞品并生成分析报告”这类复杂任务，就会暴露明显短板：

一是缺乏规划能力，拿到任务后“走一步看一步”，步骤混乱易遗漏，比如生成竞品报告时，一会儿查功能、一会儿查份额，无法形成有序流程；二是记忆过载，多步任务的中间结果、工具调用记录会撑满模型上下文，不仅增加Token成本，还会导致推理混乱（即“上下文污染”）；三是能力单一，一个智能体包揽所有工作，既做调研又做编程，结果往往样样不精，且不同任务的数据相互干扰。

而我们要搭建的DeepAgents项目，核心就是解决这三大痛点，通过“任务规划+外部记忆+子智能体协同”的架构，让智能体具备“管理能力”。接下来，我们从基础到进阶，一步步完成项目搭建。

第一步：环境准备与基础搭建，实现简单工具调用

搭建DeepAgents项目的前提，是准备好基础开发环境，核心依赖LangChain和LangGraph——这两个工具能帮我们快速实现智能体的核心逻辑，无需从零开发。

首先执行安装命令，安装所需依赖（这是项目搭建的第一步，也是后续所有功能实现的基础）：

pip install langchain langgraph openai python-dotenv  # 核心依赖，openai可替换为其他LLM库1.

安装完成后，我们从最基础的功能入手：实现一个能自主调用工具的文本处理智能体，这是DeepAgents项目的“入门操作”，也是后续复杂功能的基础。

核心思路是：用@tool装饰器将普通Python函数转换为LangChain可识别的工具，再通过create_agent函数，将LLM（大脑）、工具（能力）和系统提示词（行为准则）封装成可执行的智能体。关键代码如下：

from langchain.agents import create_agent, AgentExecutor
from langchain.agents.tools import tool
from langchain_openai import ChatOpenAI
from dotenv import load_dotenv
import os

# 加载环境变量（存储LLM API Key）
load_dotenv()
llm = ChatOpenAI(model="gpt-3.5-turbo", api_key=os.getenv("OPENAI_API_KEY"))

# 1. 用@tool装饰器定义基础工具
@tool
def count_words(text: str) -> int:
    """统计输入文本的字数，返回整数结果。"""
    return len(text.strip().split())

@tool
def convert_case(text: str, case: str = "upper") -> str:
    """将文本转换为大写或小写，case参数可选'upper'（默认）或'lower'。"""
    if case == "upper":
        return text.upper()
    elif case == "lower":
        return text.lower()
    else:
        return "参数错误，case只能为'upper'或'lower'"

# 2. 组装工具列表
tools = [count_words, convert_case]

# 3. 系统提示词（指导智能体行为）
system_prompt = """你是一个文本处理智能体，可调用提供的工具完成任务，严格按照工具参数要求调用，无需额外多余操作。"""

# 4. 构建智能体并执行
agent = create_agent(llm, tools, system_prompt)
agent_executor = AgentExecutor(agent=agent, verbose=True)

# 测试调用
result = agent_executor.invoke({"input": "统计'Hello DeepAgents'的字数，并将其转换为大写"})
print(result["output"])1.
2.
3.
4.
5.
6.
7.
8.
9.
10.
11.
12.
13.
14.
15.
16.
17.
18.
19.
20.
21.
22.
23.
24.
25.
26.
27.
28.
29.
30.
31.
32.
33.
34.
35.
36.
37.
38.
39.

比如，我们可以定义统计字数、转换大小写、提取关键词三个基础工具，用@tool装饰器标注后，LLM会自动识别工具的用途和参数。之后通过create_agent函数组装智能体，调用.invoke()方法即可运行——当用户输入“提取这段文字的关键词并转换为大写”时，智能体能自主选择对应的工具，完成任务后返回结果。

这一步的核心是掌握“工具定义”和“智能体构建”的基本逻辑：@tool装饰器的关键是清晰的文档字符串（docstring），它直接决定LLM能否正确调用工具；create_agent函数则是智能体的“组装器”，三大核心组件（LLM、Tools、System Prompt）缺一不可。

第二步：添加状态管理，让智能体拥有“记忆”

基础工具调用实现后，我们需要给智能体添加“记忆”——让它能记住任务列表和操作历史，这是实现复杂任务规划的关键。这一步，我们需要自定义智能体状态，解决普通智能体“记不住事”的痛点。

具体实现的核心，是继承LangGraph的基础AgentState，扩展自定义状态字段。我们可以定义一个状态类，继承AgentState（默认包含对话历史messages字段），再添加todos（任务列表）和files（虚拟文件系统，为后续进阶功能预留）两个字段。关键代码如下：

from langgraph.graph import AgentState
from typing import List, Dict, Annotated
from langchain_core.messages import BaseMessage
from langgraph.utils import Reducer

# 1. 自定义Reducer函数，合并任务历史
def merge_history(left: List[str], right: List[str]) -> List[str]:
    """合并操作历史，避免覆盖"""
    return left + right

# 2. 扩展智能体状态
class DeepAgentState(AgentState):
    # 任务列表：可选字段，每个任务包含content（描述）和status（状态）
    todos: List[Dict[str, str]] = []
    # 虚拟文件系统：用Reducer合并，避免覆盖文件
    files: Annotated[Dict[str, str], Reducer(merge_history)] = {}
    # 操作历史：记录所有工具调用记录
    history: Annotated[List[str], Reducer(merge_history)] = []1.
2.
3.
4.
5.
6.
7.
8.
9.
10.
11.
12.
13.
14.
15.
16.
17.
18.

其中，todos字段用于存储任务列表，我们可以用字典列表的形式，每个字典包含任务描述（content）和任务状态（status，可选待处理、进行中、已完成）；为了让任务状态能持续累积，我们会用Annotated标注字段，并搭配Reducer函数——当不同节点更新任务列表时，不会直接覆盖，而是合并新旧内容。

同时，我们需要实现两个核心工具：write_todos（创建/更新任务列表）和read_todos（读取当前任务列表）。write_todos工具通过InjectedState注入当前智能体状态，验证任务结构后，通过Command对象更新状态；read_todos工具则只读取状态，不作修改，直接返回任务列表。关键代码如下：

from langgraph.graph import InjectedState, Command

# 1. 写入/更新任务列表工具
@tool
def write_todos(todos: List[Dict[str, str]], state: InjectedState[DeepAgentState]) -> Command:
    """创建或全量更新任务列表，每个任务需包含content和status字段（status可选pending/in_progress/completed）"""
    # 验证任务结构
    for todo in todos:
        if not all(key in todo for key in ["content", "status"]):
            return Command(update={"messages": [f"任务格式错误，需包含content和status字段"]})
        if todo["status"] not in ["pending", "in_progress", "completed"]:
            return Command(update={"messages": [f"状态错误，只能是pending/in_progress/completed"]})
    # 更新状态：全量覆盖todos，追加操作历史
    return Command(
        update={
            "todos": todos,
            "history": [f"更新任务列表：{todos}"],
            "messages": [f"任务列表已更新，共{len(todos)}个任务"]
        }
    )

# 2. 读取任务列表工具
@tool
def read_todos(state: InjectedState[DeepAgentState]) -> List[Dict[str, str]]:
    """读取当前所有任务列表，返回任务详情"""
    return state.get("todos", [])

# 测试工具（需结合智能体状态）
test_state = DeepAgentState(todos=[{"content": "搜索MCP的定义", "status": "pending"}])
print(read_todos(test_state))  # 输出：[{"content": "搜索MCP的定义", "status": "pending"}]1.
2.
3.
4.
5.
6.
7.
8.
9.
10.
11.
12.
13.
14.
15.
16.
17.
18.
19.
20.
21.
22.
23.
24.
25.
26.
27.
28.
29.
30.

举个例子，当用户输入“调研MCP并总结”时，智能体会通过write_todos工具生成任务列表（“搜索MCP的定义”“整理MCP的核心功能”“总结MCP的应用场景”），并实时更新任务状态；执行过程中，智能体可以通过read_todos工具查看当前进度，确保任务有序推进。这一步，我们正式解决了普通智能体“缺乏规划”的痛点。

第三步：实现并行工具调用，提升多任务处理效率

当智能体拥有“记忆”后，我们可以进一步优化效率——实现并行工具调用，让智能体能够同时处理多个独立任务，解决普通智能体“串行执行效率低”的问题。

这一步的实现无需额外新增复杂工具，核心是利用LangGraph的原生能力：当LLM判断多个任务之间没有依赖关系时，会在一次响应中生成多个tool_call，而create_agent函数能自动识别这种情况，并行执行这些工具调用，而非串行等待。

比如，用户输入“同时查询北京和上海的天气，以及美元和欧元的汇率”，智能体无需先查完北京天气，再查上海天气，而是会并行发起四个查询请求，同时获取结果，大幅缩短任务执行时间。

这里需要注意的是，并行调用的关键的是LLM的决策能力——我们需要在系统提示词中明确指导模型，当遇到多个独立任务时，可发起多个工具调用。到这一步，我们的DeepAgents项目已经具备了“规划能力”和“高效执行能力”，能应对中等复杂度的任务。

第四步：引入虚拟文件系统，实现上下文卸载

随着任务复杂度提升，我们会遇到新的问题：处理大文件、长日志时，大量内容会撑爆LLM的上下文窗口，导致任务失败。这时候，我们需要给项目添加“虚拟文件系统”，实现上下文卸载，解决普通智能体“记忆过载”的痛点。

上下文卸载的核心逻辑很简单：不再把所有中间结果、大篇幅内容塞进对话上下文，而是将其写入虚拟文件系统，对话历史中只保留“文件路径+预览”，需要时再通过工具读取。这就像我们整理电脑文件，把大文件存入硬盘，桌面只留快捷方式，既节省空间，又能快速调取。

具体实现上，我们基于之前定义的DeepAgentState，完善files字段（字典类型，key为文件路径，value为文件内容），并实现三个核心工具：ls（列出所有文件路径）、read_file（分页读取文件内容）、write_file（写入/覆盖文件）。关键代码如下：

from langgraph.graph import InjectedState, Command

# 1. 列出所有文件（ls工具）
@tool
def ls(state: InjectedState[DeepAgentState]) -> List[str]:
    """列出虚拟文件系统中所有文件的路径"""
    files = state.get("files", {})
    return list(files.keys()) if files else ["当前无任何文件"]

# 2. 分页读取文件（read_file工具）
@tool
def read_file(file_path: str, offset: int = 0, limit: int = 10, state: InjectedState[DeepAgentState]) -> str:
    """分页读取文件内容，offset为起始行号，limit为最大读取行数"""
    files = state.get("files", {})
    if file_path not in files:
        return f"文件不存在：{file_path}"
    # 按行分割，分页返回
    lines = files[file_path].splitlines()
    end = offset + limit
    paginated_lines = lines[offset:end]
    # 添加行号，便于查看
    return "n".join([f"{offset+i+1}: {line}" for i, line in enumerate(paginated_lines)])

# 3. 写入/覆盖文件（write_file工具）
@tool
def write_file(file_path: str, content: str, state: InjectedState[DeepAgentState]) -> Command:
    """写入或覆盖文件，file_path为文件路径，content为文件内容"""
    current_files = state.get("files", {})
    # 覆盖或新增文件
    current_files[file_path] = content
    return Command(
        update={
            "files": current_files,
            "history": [f"写入文件：{file_path}，内容长度：{len(content)}字符"],
            "messages": [f"文件{file_path}已成功写入"]
        }
    )

# 测试文件工具
test_state = DeepAgentState(files={})
# 写入文件
write_cmd = write_file("search_results.txt", "MCP是智能体相关的核心概念...", test_state)
test_state.update(write_cmd.update)
# 列出文件
print(ls(test_state))  # 输出：["search_results.txt"]
# 读取文件
print(read_file("search_results.txt", 0, 5, test_state))1.
2.
3.
4.
5.
6.
7.
8.
9.
10.
11.
12.
13.
14.
15.
16.
17.
18.
19.
20.
21.
22.
23.
24.
25.
26.
27.
28.
29.
30.
31.
32.
33.
34.
35.
36.
37.
38.
39.
40.
41.
42.
43.
44.
45.
46.
47.

其中，read_file工具支持offset（起始行号）和limit（读取行数）参数，避免一次性读取大文件撑爆上下文；write_file工具通过Command对象更新文件系统，并在对话历史中追加文件更新提示；ls工具则用于查看当前所有文件，方便智能体和用户了解文件存储情况。

比如，我们处理大篇幅的调研数据时，智能体会通过write_file工具将数据写入“search_results.txt”文件，后续分析时，通过read_file工具分页读取内容，既避免了上下文过载，又能灵活处理大文件。这一步，我们彻底解决了普通智能体“记忆过载”的痛点，让项目能应对长周期、大内容的复杂任务。

第五步：添加子智能体委托，实现专业分工协同

当项目需要处理更复杂的任务（如行业研究、多领域调研）时，单一智能体的能力仍有局限。这时候，我们需要给项目添加“子智能体委托”功能，让主智能体能够分派任务，实现专业分工，解决普通智能体“能力单一”的痛点。

核心实现分为三步：定义子智能体配置、实现task工具、组装主监督智能体。

首先，用TypedDict定义SubAgent配置，相当于给子智能体编写“岗位说明书”，包含name（唯一标识，如“research-agent”）、description（功能描述，供主智能体决策）、prompt（系统提示词，指导子智能体工作）、tools（子智能体可使用的工具，如调研类子智能体只允许使用搜索工具）。关键代码如下：

from typing import TypedDict, List
from langchain.agents import create_agent
from langchain_core.tools import BaseTool

# 1. 定义子智能体配置（TypedDict）
class SubAgent(TypedDict):
    name: str  # 子智能体唯一标识
    description: str  # 子智能体功能描述（供主智能体决策）
    prompt: str  # 子智能体系统提示词
    tools: List[BaseTool]  # 子智能体可使用的工具

# 2. 定义调研类子智能体配置（示例）
research_agent_config: SubAgent = {
    "name": "research-agent",
    "description": "专注于行业调研、信息搜索，擅长通过搜索工具获取精准信息并整理",
    "prompt": "你是调研子智能体，仅使用web_search工具获取信息，整理后以简洁明了的格式返回，不添加多余内容",
    "tools": [web_search]  # web_search为之前实现的搜索工具（可替换为Tavily API）
}

# 3. 实现create_task_tool函数，把子智能体转为可调用工具
def create_task_tool(subagents: List[SubAgent], llm) -> BaseTool:
    @tool
    def task(description: str, subagent_type: str, state: InjectedState[DeepAgentState]) -> Command:
        """
        委托子智能体执行任务，description为任务描述，subagent_type为子智能体名称（对应SubAgent的name）
        """
        # 找到指定子智能体
        subagent = next((sa for sa in subagents if sa["name"] == subagent_type), None)
        if not subagent:
            return Command(update={"messages": [f"未找到子智能体：{subagent_type}"]})
        
        # 创建隔离上下文（仅包含任务描述和文件，不包含主智能体历史）
        isolated_state = DeepAgentState(
            messages=[{"role": "user", "content": description}],
            files=state.get("files", {})
        )
        
        # 构建子智能体并执行
        sub_agent = create_agent(llm, subagent["tools"], subagent["prompt"])
        sub_agent_executor = AgentExecutor(agent=sub_agent, verbose=True)
        result = sub_agent_executor.invoke(isolated_state)
        
        # 把子智能体结果汇回主智能体
        return Command(
            update={
                "messages": [f"子智能体{subagent_type}执行结果：{result['output']}"],
                "history": [f"委托子智能体{subagent_type}执行任务：{description}"]
            }
        )
    return task1.
2.
3.
4.
5.
6.
7.
8.
9.
10.
11.
12.
13.
14.
15.
16.
17.
18.
19.
20.
21.
22.
23.
24.
25.
26.
27.
28.
29.
30.
31.
32.
33.
34.
35.
36.
37.
38.
39.
40.
41.
42.
43.
44.
45.
46.
47.
48.
49.
50.

在技术落地时，我们选择Tavily API作为搜索工具（专为AI设计，结果质量高，免费额度1000次/月，适合快速落地），并自定义行业研究框架，让智能体能够按“市场规模-发展历程-竞争格局-发展趋势”的维度，自动规划任务、调用工具、存储数据、生成报告。关键代码如下：

from tavily import TavilyClient
from dotenv import load_dotenv
import os

# 1. 初始化Tavily搜索工具
load_dotenv()
tavily = TavilyClient(api_key=os.getenv("TAVILY_API_KEY"))

@tool
def web_search(query: str) -> str:
    """使用Tavily API搜索信息，返回简洁的搜索结果摘要"""
    response = tavily.search(query, max_results=3)
    return "n".join([f"来源：{res['url']}n内容：{res['content'][:200]}..." for res in response["results"]])

# 2. 自定义行业研究框架
RESEARCH_FRAMEWORK = {
    "market_size": {
        "name": "市场规模",
        "search_queries": ["{industry}市场规模 2024-2025", "{industry}市场增长率"],
        "output_template": "# {industry}市场规模n## 市场体量n{market_volume}n## 增长率n{growth_rate}"
    },
    "competition": {
        "name": "竞争格局",
        "search_queries": ["{industry}主要企业及市场份额", "{industry}核心竞品分析"],
        "output_template": "# {industry}竞争格局n## 主要企业n{companies}n## 市场份额n{market_share}"
    }
}

# 3. 组装主智能体，落地行业研究助手
def create_industry_research_agent(llm):
    # 子智能体配置
    subagents = [research_agent_config]
    # 创建task工具
    task_tool = create_task_tool(subagents, llm)
    # 基础工具列表
    base_tools = [write_todos, read_todos, ls, read_file, write_file, task_tool, web_search]
    # 系统提示词（指导主智能体按框架调研）
    system_prompt = f"""你是AI行业研究助手，需按以下框架完成调研：{RESEARCH_FRAMEWORK}
步骤：1. 生成调研任务列表；2. 委托research-agent执行搜索；3. 写入文件；4. 整合生成报告。"""
    # 构建主智能体
    agent = create_agent(llm, base_tools, system_prompt, state_schema=DeepAgentState)
    return AgentExecutor(agent=agent, verbose=True)

# 测试行业研究助手
agent_executor = create_industry_research_agent(llm)
result = agent_executor.invoke({"input": "调研AI生成式内容行业"})
print(result["output"])1.
2.
3.
4.
5.
6.
7.
8.
9.
10.
11.
12.
13.
14.
15.
16.
17.
18.
19.
20.
21.
22.
23.
24.
25.
26.
27.
28.
29.
30.
31.
32.
33.
34.
35.
36.
37.
38.
39.
40.
41.
42.
43.
44.
45.
46.
47.

其次，通过create_task_tool函数，把子智能体变成可调用的工具。这个工具的核心作用是：主智能体调用task(description, subagent_type)时，会创建一个隔离的上下文（只包含任务描述和必要文件，不包含主智能体的完整对话历史），启动对应的子智能体执行任务，完成后将结果封装成ToolMessage汇回主智能体。

最后，通过create_supervisor_agent函数组装主智能体，将task工具和其他基础工具注入，主智能体负责制定战略、拆解任务、验收子智能体的结果，不直接执行具体操作。

举个例子，当主智能体需要完成“AI行业研究报告”时，会将任务拆分为“市场规模调研”“竞争格局分析”“发展趋势预测”三个子任务，分别委托给三个不同的子智能体（research-agent），三个子智能体并行执行，完成后将结果汇总给主智能体，主智能体再整合生成最终报告。这种模式既保证了专业性，又提升了效率。

此外，我们还可以添加一个think_tool工具，让主智能体在拿到子智能体结果后，显式记录思考过程，反思已掌握的信息、缺失的内容，规划下一步动作，确保任务整体可控。

项目落地：搭建AI行业研究助手，检验DeepAgents能力

当我们完成了以上五步搭建后，就可以将项目落地为一个实用的AI行业研究助手，检验DeepAgents的核心能力。这个助手适用于投资人、产品经理、求职者等各类人群，能帮助用户快速了解陌生行业。

整体架构如下：

在技术落地时，我们选择Tavily API作为搜索工具（专为AI设计，结果质量高，免费额度1000次/月，适合快速落地），并自定义行业研究框架，让智能体能够按“市场规模-发展历程-竞争格局-发展趋势”的维度，自动规划任务、调用工具、存储数据、生成报告。

具体流程如下：用户输入目标行业（如“AI生成式内容行业”），主智能体通过write_todos工具生成调研任务列表；之后，主智能体委托子智能体，通过Tavily API搜索各维度信息，将搜索结果通过write_file工具存入虚拟文件系统；最后，主智能体整合文件中的信息，生成结构清晰的行业研究报告，全程无需用户手动干预。

同时，我们可以添加流式输出功能，让用户实时看到任务进度和报告生成过程；通过状态同步函数，实现任务状态的实时更新，用户随时可以查看当前调研进度。

关键技术选型

组件	技术选择	选择理由
大模型	DeepSeek	中文能力强，推理成本低，适合复杂任务规划。
搜索引擎	Tavily API	专为 AI 设计的搜索 API，结果质量高，免费额度 1000 次/月，适合快速落地。
智能体框架	LangGraph	提供状态管理和流程控制，支持复杂工作流编排。
Web 框架	Streamlit	快速构建数据应用，自带响应式布局和组件系统。
状态管理	LangGraph State	通过状态注解实现跨工具的状态共享和持久化。

核心功能实现

智能体首先使用 write_todos 工具将复杂的研究任务分解为可执行的子任务：

# 用户输入：新能源汽车行业
# 智能体自动生成的任务列表
todos = [
    {"content": "研究新能源汽车行业概述", "status": "pending"},
    {"content": "研究新能源汽车竞争格局", "status": "pending"},
    {"content": "研究新能源汽车发展趋势", "status": "pending"},
    {"content": "整合分析并生成报告", "status": "pending"}
]1.
2.
3.
4.
5.
6.
7.
8.

每个任务都有明确的状态：pending（待处理）、in_progress（进行中）、completed（已完成），通过 update_todo_status 工具实时更新。这种结构化的任务管理方式，让用户能够清晰看到研究进度。

智能体根据任务列表，自动进行多维度的信息收集：

# 搜索工具调用示例
web_search("新能源汽车市场规模和增长趋势")
web_search("新能源汽车主要企业和竞争格局")
web_search("新能源汽车技术发展趋势")
web_search("新能源汽车政策环境和未来展望")1.
2.
3.
4.
5.

每次搜索后，智能体会使用 think 工具进行反思：

“我找到了什么关键信息？”
“这些信息是否足够回答用户的问题？”
“还需要搜索哪些补充信息？”

这种反思-决策机制，确保了搜索的针对性和信息收集的完整性。

所有研究结果被实时存储在虚拟文件系统中，通过 write_file 工具写入：

# 文件写入示例
write_file("行业概述.md", "# 新能源汽车行业概述n...")
write_file("竞争格局.md", "# 新能源汽车竞争格局n...")
write_file("发展趋势.md", "# 新能源汽车发展趋势n...")1.
2.
3.
4.

这个虚拟文件系统有以下优点：

隔离性：每次研究都有独立的文件空间，避免数据混乱。
持久性：通过 LangGraph 的状态管理，文件在研究过程中持续保存。
可扩展：支持任意数量和类型的文件，适合存储不同维度的研究结果。

最后，主智能体整合所有文件中的信息，生成结构清晰的行业研究报告：

# 最终报告生成
report = generate_report(
    overview=read_file("行业概述.md"),
    competition=read_file("竞争格局.md"),
    trends=read_file("发展趋势.md")
)
write_file("研究报告.md", report)1.
2.
3.
4.
5.
6.
7.

报告自动包含摘要、核心发现、详细分析、结论建议等部分，完全符合专业研究报告的格式要求。

关键代码解析

我们扩展了 LangGraph 的 AgentState，增加了任务列表和文件系统：

class DeepAgentState(AgentState):
    """
    深度智能体状态
    包含：
    - messages: 消息历史（继承自 AgentState）
    - todos: TODO 任务列表
    - files: 虚拟文件系统
    """
    todos: NotRequired[list[dict]]
    files: Annotated[NotRequired[dict[str, str]], file_reducer]1.
2.
3.
4.
5.
6.
7.
8.
9.
10.

这里的 file_reducer 是一个自定义的合并函数，用于在状态更新时正确合并新旧文件：

def file_reducer(left: dict, right: dict) -> dict:
    """文件系统合并函数"""
    if left is None:
        return right
    elif right is None:
        return left
    else:
        return {**left, **right}  # 新文件覆盖旧文件1.
2.
3.
4.
5.
6.
7.
8.

通过 LangGraph 的状态注入机制，工具可以安全地访问和修改状态：

@tool
def write_file(
    file_path: str,
    content: str,
    state: Annotated[dict, InjectedState] = None,  # 注入状态
    tool_call_id: Annotated[str, InjectedToolCallId] = None,  # 注入工具调用ID
) -> Command:
    """向虚拟文件系统写入内容"""
    files = state.get("files", {})
    files[file_path] = content

    # 返回 Command 对象更新状态
    return Command(
        update={
            "files": files,
            "messages": [
                ToolMessage(
                    content=f"已成功写入文件: {file_path}",
                    tool_call_id=tool_call_id,
                )
            ],
        }
    )1.
2.
3.
4.
5.
6.
7.
8.
9.
10.
11.
12.
13.
14.
15.
16.
17.
18.
19.
20.
21.
22.
23.

这种设计保证了工具之间的解耦和状态的统一管理。

智能体的创建过程清晰地体现了 DeepAgents 的设计理念：

def create_research_agent(
    model_name: str = None,
    temperature: float = None,
    max_results: int = None,
    search_depth: str = None,
):
    """创建行业研究智能体"""

    # 1. 初始化大模型
    llm = ChatDeepSeek(
        model=actual_model_name,
        temperature=actual_temperature,
        api_key=settings.DEEPSEEK_API_KEY,
    )

    # 2. 创建工具集
    tools = [
        web_search,      # 搜索工具
        ls, read_file, write_file,  # 文件工具
        write_todos, read_todos, update_todo_status,  # 任务工具
        think,           # 思考工具
    ]

    # 3. 创建智能体
    agent = create_agent(
        llm,
        tools=tools,
        system_prompt=SYSTEM_PROMPT,
        state_schema=DeepAgentState,
    )

    return agent1.
2.
3.
4.
5.
6.
7.
8.
9.
10.
11.
12.
13.
14.
15.
16.
17.
18.
19.
20.
21.
22.
23.
24.
25.
26.
27.
28.
29.
30.
31.
32.

项目亮点

for chunk in agent.stream(
    {"messages": [HumanMessage(content=prompt)]},
    config=config,
    stream_mode="values",
):
    # 实时更新进度和日志
    if "todos" in chunk:
        st.session_state.todos = chunk["todos"]
        update_progress(calculate_progress(chunk["todos"]))1.
2.
3.
4.
5.
6.
7.
8.
9.

这种实时反馈机制，大大提升了用户体验，让用户能够信任智能体的工作过程。

项目实现了完善的状态同步和容错机制：

def run_real_research(industry: str):
    """使用真实智能体进行研究"""
    try:
        # 尝试使用真实智能体
        agent = create_research_agent()
        # ... 智能体运行逻辑
    except Exception as e:
        # 出错时自动切换到模拟模式
        add_log("info", f"❌ 研究出错: {str(e)}")
        add_log("info", "切换到模拟模式...")
        simulate_research(industry)1.
2.
3.
4.
5.
6.
7.
8.
9.
10.
11.

这种设计保证了服务的稳定性，即使在 API 限流或网络故障的情况下，用户仍然可以得到模拟结果。

项目支持研究历史记录和报告导出：

# 保存到历史记录
history_record = {
    "industry": industry,
    "time": datetime.now().strftime("%Y-%m-%d %H:%M:%S"),
    "files": dict(st.session_state.research_files),
    "result": st.session_state.research_result,
    "todos": list(st.session_state.todos),
}
st.session_state.research_history.append(history_record)

# 只保留最近 10 条历史
if len(st.session_state.research_history) > 10:
    st.session_state.research_history = st.session_state.research_history[-10:]1.
2.
3.
4.
5.
6.
7.
8.
9.
10.
11.
12.
13.

项目价值与能力检验

验证了 DeepAgents 的几项核心能力：

能力维度	验证方式	结果体现
任务规划能力	自动生成结构化任务列表	从“新能源汽车”自动分解为概述、格局、趋势、报告四个任务
工具协调能力	多工具协同完成复杂任务	搜索→思考→存储→报告的完整流程自动化
状态管理能力	维护多维度中间状态	任务进度、文件系统、研究日志的一致性管理
错误恢复能力	异常情况下的降级服务	API 故障时自动切换模拟模式，保证用户体验
用户体验	流式输出与实时反馈	用户能看到智能体的思考过程，建立信任感

总结：循序渐进，从入门到精通DeepAgents项目搭建

我们搭建DeepAgents项目的过程，正是解决普通智能体痛点的过程：从基础的工具调用，到状态管理实现任务规划，再到并行调用提升效率，最后通过虚拟文件系统和子智能体委托，实现复杂任务的高效处理，每一步都循序渐进、可落地。

整个项目的核心逻辑，是让智能体从“被动执行”升级为“主动管理”——通过“规划-分解-委托-整合”的闭环架构，让智能体能够调度资源、管理记忆、分工协作。对于开发者而言，借助LangChain和LangGraph，我们无需从零开发核心逻辑，只需按步骤完善功能，就能快速搭建出具备DeepAgents能力的智能体。

未来，我们还可以基于这个项目进一步优化：将虚拟文件系统与持久化存储（如Postgres）结合，实现跨会话的长期记忆；优化子智能体的配置，支持更多专业场景；添加UI界面，让非技术用户也能轻松使用。相信随着技术的迭代，DeepAgents将在更多场景中落地，成为我们工作中的“智能伙伴”。

作者介绍

崔皓，51CTO社区编辑，资深架构师，拥有18年的软件开发和架构经验，10年分布式架构经验。

手把手搭建DeepAgents项目：从基础到进阶