利用LangChain Deep Agents重新构建类似Claude的代码风格代理。

编程智能体并非聊天机器人

人们很容易将编程智能体想象成更智能的聊天机器人，但这种看法是错误的。Claude Code的能力并非仅来自模型本身，而是来自围绕该模型构建的智能体框架。这个框架就是控制层：一个看不见的系统，它为模型提供读取文件的工具、运行代码的环境、记录已执行操作的历史，以及确保任务按计划推进的检查清单。如果没有这个框架，语言模型就只是个文本生成器；而有了它，同一个模型就能表现得像一个能力出众的自主软件工程师。

我们的计划是逐个分析Claude Code风格智能体的工作原理，并使用LangChain的deepagents库重新构建每个组成部分。到最后，当你亲自构建出一个这样的智能体时，它的结构就会变得清晰易懂。

核心的循环机制

会话从用户的消息开始。这可能是一个关于文件夹内容的简单问题，也可能是一项艰巨的任务，比如重构登录系统并更新相关测试。模型会在选择下一步行动前先读取该消息，之后路径就会分岔。当需要执行操作时，它会请求使用某个工具，得到输出结果后再次循环处理，这一流程对于一步任务和五十步任务来说是完全相同的。一旦模型用普通文字回复且不再请求工具，循环就会结束，用户就能看到结果。没有工具调用这一情况正是表明智能体认为任务已经完成的信号。

“决定下一步”框整合了代理研究中所描述的三个阶段：通过读取文件收集上下文、通过编辑和执行命令来采取行动，以及通过运行测试进行验证。这些阶段并非以独立模式运行，而只是在同一循环中依次执行的不同工具调用而已。

还有一个刻意设计的模式转变，即计划模式。通常情况下，代理会同时进行规划和行动。在计划模式下，循环之前会有一个判断节点。代理会在不修改任何内容的情况下进行调研并起草方案：文件会被读取，思路也会被梳理，但编辑和命令操作要等到用户确认后才会执行，只有此时正常的循环才会继续。这为处理大型或高风险任务提供了安全保障。它不同于代理为每项任务自行维护的任务清单，因为计划模式是由用户触发且必须经过明确批准的暂停状态。

该图仍然没有展示围绕在循环之外的各种机制：内存满时的压缩处理、运行各自独立循环的子代理，以及介于“执行工具”与实际系统之间的权限与沙箱层。这些机制并不会改变循环本身，只是环绕在它周围。

控制框架内包含四种功能，正是它们维持着循环的稳定运行。规划功能让代理先进行思考；工具则充当其“手”的角色；上下文管理功能可防止耗时较长的任务占用过多内存；子代理则将辅助任务放入各自的独立循环中，从而避免主线程变得混乱。它们为循环提供支持，但从不重写循环本身。

在所有这些层之下，存在着最为重要的那一层。当智能体想要通过编辑文件、运行命令或使用网络来接触现实世界时，其操作并不会直接到达机器。这些操作首先会经过权限检查与沙箱机制的过滤，而在整个设计中，没有其他安全措施比这更为关键。试想，若想在智能体的指令中加入“请不要删除任何重要内容”这样的提示来确保其安全，那不过是一张便签而已——模型可能会忽略它、误解它，或者通过争论绕过它。真正的保护措施存在于模型之外：有一条会直接拒绝工具调用的权限规则，还有一个操作系统级的沙箱机制，无论模型如何选择，都会阻止危险的命令执行。

纯粹的引擎

在引入任何库之前，先了解没有附加任何功能的引擎结构会更有帮助。这是一个完整且可运行的智能体循环。

def run_agent_loop(client, user_message, tools, tool_functions, max_turns=20):
    messages = [{"role": "user", "content": user_message}]
    for _ in range(max_turns):
        # 1. Ask the model what to do next.
        response = client.messages.create(
            model="claude-sonnet-4-6", max_tokens=1024, tools=tools, messages=messages
        )
        messages = [*messages, {"role": "assistant", "content": response.content}]

        # 2. Plain text and no tool request? The job is done.
        tool_uses = [b for b in response.content if b.type == "tool_use"]
        if not tool_uses:
            return "".join(b.text for b in response.content if b.type == "text")
        # 3. Run each requested tool and hand the results back. 4. Repeat.
        results = [
            {"type": "tool_result", "tool_use_id": call.id,
             "content": tool_functions[call.name](**call.input)}
            for call in tool_uses
        ]
        messages = [*messages, {"role": "user", "content": results}]
    raise RuntimeError(f"agent loop did not finish within {max_turns} turns")

代码总量大约在三十行左右，其余部分则是带有附加功能的循环结构。功能更强的工具会被放入tool_functions中；规划器会在messages里记录笔记；当messages内容过多时，压缩器会对其进行整理。子代理其实就是再次调用相同的功能，只不过使用的是一个新的、空的messages列表。权限检查则安排在模型请求工具与执行该工具的代码行之间。max_turns限制则是所有实际应用系统都会设置的预防措施，因为模型如果陷入循环，就会无止境地消耗资源。

每个编程代理都有一个核心机制，也就是将指令转化为实际操作的代理循环，其结构简单到令人惊讶：

1. 询问模型下一步该做什么。
2. 模型要么以纯文本形式回复，要么请求使用某个工具。

如果系统请求使用某个工具，它会运行该工具并将结果返回给模型。

回到第一步。

当模型以纯文本形式回复且无需使用工具时，任务即完成，此时应停止并显示答案。

读取文件、执行命令以及编辑代码都需经历这个循环过程。可以想象一名承包商接到任务后有一份清单：查看下一步该做什么、执行操作、检查结果、再次选择，直到清单为空才汇报进展。这个循环就起到了该承包商决策的作用。此外它还具有自适应能力，因为简单的问题可能只需要一次循环，而大规模代码重构则需要数十次工具调用，而且无需事先确定循环次数。

利用深度智能体构建它

无需手动编写 while 循环。创建 Deep Agent 即可自动提供可用的循环结构。首先需要安装相关库和模型提供器。

pip install deepagents langchain-anthropic

然后创建最简单的智能体。

from deepagents import create_deep_agent

def get_weather(city: str) -> str:
    """Get the weather for a given city."""
    return f"It's always sunny in {city}!"
agent = create_deep_agent(
    model="anthropic:claude-sonnet-4-6",
    tools=[get_weather],
    system_prompt="You are a helpful assistant.",
)
result = agent.invoke(
    {"messages": [{"role": "user", "content": "What's the weather in San Francisco?"}]}
)

create_deep_agent函数调用会返回一个已内置循环结构的智能体，而.invoke()则会执行上述的循环流程。该框架并不关心底层使用的是哪种模型，无论是OpenAI、Google还是Anthropic，循环结构都保持不变。

工具就是双手

如果说循环是引擎，那么工具就是双手。没有工具的话，模型只能进行文本交流；而有了好的工具，它就能读取代码库、对其进行修改并验证结果。真正的编程工具有四类：

• 读取工具用于在不做任何修改的情况下查看内容：read_file用于打开文件，glob可根据名称模式查找文件，比如所有.py文件，而grep则用于在文件中搜索特定文本。初学者常常将后两者混淆，但实际上glob是按名称查找文件，而grep则是按内容查找。

编辑工具能安全地修改内容：write_file用于创建新文件，而edit_file则用于修改现有文件的部分内容。

执行工具可运行shell命令，即那些在终端中输入的命令。

委托工具task会将子任务交给辅助代理处理。

为何不让代理使用cat来读取内容、用sed来进行编辑呢？虽然也可以这样做，但专为特定功能设计的工具在两方面更具优势。其一在于令牌预算管理：专用的读取工具可以在加载文件前先查看其大小，必要时进行压缩，而原始的cat命令则会将整个文件内容一次性输出，可能导致上下文窗口超出限制。其二在于更清晰的权限控制：调用已知的edit_file工具时，框架能够准确了解正在发生什么，从而应用相应的安全规则；而随意编写的shell命令则具有不可预测性且风险较高。

这里有个有趣的怪现象：这些模型是从充斥着大量使用 cat 和 sed 的 Stack Overflow 回答的公共网络中学习而来的，因此即便是最先进的模型也会出于习惯而使用原始 shell 命令。系统提示必须不断提醒它们使用真正的工具，而非 shell 的快捷命令。

Deep Agents 内置了完整的功能集：规划、文件系统工具、执行以及任务委托功能。自定义工具则作为装饰函数添加进来。

import subprocess
from langchain_core.tools import tool

MAX_OUTPUT_CHARS = 20_000  # roughly 5k tokens
@tool
def run_tests(path: str = ".") -> str:
    """Run the project's pytest suite and return its output.
    Output is truncated to the last 20k characters (failures appear at the
    end). The run is killed after 5 minutes.
    """
    try:
        result = subprocess.run(["pytest", path], capture_output=True,
                                text=True, check=False, timeout=300)
    except subprocess.TimeoutExpired:
        return "pytest timed out after 300s"
    output = result.stdout + result.stderr
    if len(output) > MAX_OUTPUT_CHARS:
        return ("[... output truncated to the last 20,000 characters ...]\n"
                + output[-MAX_OUTPUT_CHARS:])
    return output
agent = create_deep_agent(
    model="anthropic:claude-sonnet-4-6",
    tools=[run_tests],            # merged in alongside the built-ins
    system_prompt="You are a coding assistant. Always run tests after editing.",
)

@tool装饰器能让普通函数被模型调用。其文档字符串非常重要，因为模型会将其视为说明何时以及如何使用该工具的指南。

行动前的规划

将复杂的请求交给未经处理的模型时，它往往会出现混乱：这里调用一次，那里调用一次，没有整体规划，有时还会陷入循环。优秀的编程代理会在动手之前先制定计划，它们会编写一份结构化的待办清单，将任务分解为明确的步骤，然后按顺序执行，并标记每项任务的进展状态或完成情况。这就好比厨师在打开炉灶之前会先阅读完整份食谱，这样在第三步还未完成时就不会临时想出第四步的内容。

其中有一个巧妙的工程技巧。在长时间的会话中，最初的目标会被大量的工具输出信息所掩盖，模型也会因此忘记它。为了解决这个问题，该系统会在工具调用之后将当前的待办列表重新放入对话中，就像同事把清单推回桌面以作为提醒一样。

在 Deep Agents 中，这就是内置的 write_todos 工具。无需进行任何配置，因为默认的系统提示已经教会了智能体在行动前先制定计划，并在循环过程中持续更新待办列表。

上下文管理

最棘手的工程难题就在这里。每个模型都有一个上下文窗口，即工作内存的固定容量上限。少量大文件加上几条指令就会很快填满它，一旦超出限制，智能体就会开始丢弃内容，甚至包括最初给出的指令。编程类智能体采用两种策略来应对这一问题：

1. 将文件作为外部内存使用。智能体不会把庞大的搜索结果直接放入聊天内容中，而是将其写入磁盘，仅保留文件名，需要详细信息时再重新打开该文件。
2. 当对话内容接近容量极限时，系统会触发压缩机制。智能体会停止当前循环，整合较早的对话内容，将关键事实存储到长期记忆中，并清除已膨胀的短期记忆。

想象一下要撰写一份庞大的报告：没人能记住所有的内容。你会做笔记、整理资料，并在桌上放一份一页的摘要。Deep Agents 已经内置了这两项功能——它提供虚拟文件系统用于存储大量结果，同时还具备自动摘要生成中间件，因此从代理创建的那一刻起就具备了这样的内存管理能力。

子代理：分而治之

有些任务过于复杂，无法在单次对话中完成。搜索庞大的代码库可能意味着需要读取五十个文件，但五十个文件不应占用主智能体的可用内存。解决方法是使用子智能体。主智能体会启动一个辅助智能体，赋予其特定的任务，比如定位认证逻辑所在的位置，该辅助智能体则在其独立的空白上下文窗口中工作。完成任务后，它会返回简洁的总结。就像主管将研究任务分配给初级员工并接收一段总结一样，这样就能保持工作秩序井然。

此外还有安全机制：子智能体被严格禁止再创建自己的子智能体。否则，混乱的智能体可能会生成无止境的嵌套辅助智能体链，导致进程失控并耗尽API配额。严格的深度限制有助于确保任务分配的安全性。

Deep Agents 内置了用于生成辅助工具的 task 工具。诸如专用代码搜索器之类的特殊工具，则通过简单的字典来描述。

code_searcher = {
    "name": "code-searcher",
    "description": "Searches the codebase to find where specific logic lives. "
                   "Use this for any open-ended 'where is X?' question.",
    # ⚠️ NOTE: the key is `system_prompt`, NOT `prompt` (see correction below).
    "system_prompt": "You are an expert at navigating codebases. Use the grep "
                     "and glob tools to locate relevant files, then report a "
                     "concise summary of what you found and where. Do not make "
                     "any edits.",
}

agent = create_deep_agent(
    model="anthropic:claude-sonnet-4-6",
    tools=[run_tests],
    system_prompt="You are a coding assistant.",
    subagents=[code_searcher],
)

主代理现在知道有专门工具存在，这样它的记忆空间就能为真正的编码工作腾出空间。

安全性与人工审批

在此阶段，该代理可以编辑文件并在真实机器上运行shell命令，这一点值得深思。过于积极的代理可能会删除错误的文件或执行破坏性操作。没有约束的力量实则是一种隐患。编码代理通过两层机制来加以控制：

1. 允许列表和禁止规则能让安全的操作（如读取文件）自动执行，而具有破坏性的操作（如删除文件）则需经过审批。
2. 审批提示，也称为人工干预机制，会让代理在执行高风险操作前暂停并请求确认。用户可以批准、修改或拒绝该操作，之后流程才会继续。

智能体安全的核心原则是：提示语不能成为安全屏障。要求模型不要删除任何重要内容只是一种礼貌的请求，并非真正的障碍，因为模型仍可能绕过这一要求而产生错误输出。强制性的安全措施必须存在于模型之外，即在控制系统中实现。权限规则可以拒绝工具调用，沙箱则能在操作系统层面阻止任何危险的命令执行，无论模型多么试图让其运行。正因如此，执行工具需要一个合适的后端支持。礼貌并不等同于安全。

Deep Agents将其限制条件设置在控制系统中。访问shell环境需要后端支持，而LangGraph内置的中断机制则会在执行高风险操作之前暂停程序运行。

from deepagents import create_deep_agent
# Import path for backends can vary by version - check the current
# "Backends" page in the Deep Agents docs.
from deepagents.backends import LocalShellBackend

agent = create_deep_agent(
    model="anthropic:claude-sonnet-4-6",
    system_prompt="You are a coding assistant working inside this project.",
    backend=LocalShellBackend(),   # enables the `execute` shell tool
)

如果请求使用受限制的工具，循环将会停止，结果中会包含一个__interrupt__载荷，其中列明了待处理的操作以及可选的选项：approve、edit、reject或respond。做出决策后，程序会继续运行。

from langgraph.types import Command

result = agent.invoke({"messages": [...]}, config)
result["__interrupt__"]   # the pending action + allowed decisions — show your user

# You decide; the loop picks up exactly where it paused:
agent.invoke(Command(resume={"decisions": [{"type": "approve"}]}), config)
# ... or {"type": "reject"} — the tool is never run, and the model is told so.

真正的限制应存在于控制机制中，而不仅仅体现在提示语里。

会话间的记忆

默认情况下，一旦对话结束，智能体就会丢失所有信息。一个真正有用的助手应当记住项目的编码规范、用户的偏好以及昨天停下的工作位置。有两种机制可以实现这一功能：检查点功能能够保存智能体的精确状态，这样长时间的任务在被打断后仍能从相同位置继续；长期记忆则用于存储那些不会因单次对话而消失的信息，比如某个项目需要用四个空格缩进。

由于 Deep Agents 基于 LangGraph 运行，持久化功能是通过接入检查点工具来实现的。

from deepagents import create_deep_agent
from langgraph.checkpoint.memory import InMemorySaver

agent = create_deep_agent(
    model="anthropic:claude-sonnet-4-6",
    tools=[run_tests],
    system_prompt="You are a coding assistant.",
    checkpointer=InMemorySaver(),   # remembers state within a session
)
# A "thread_id" ties messages together into one ongoing conversation.
config = {"configurable": {"thread_id": "project-alpha"}}
agent.invoke(
    {"messages": [{"role": "user", "content": "Start refactoring the auth module."}]},
    config=config,
)
# Later, same thread_id - the agent remembers the earlier turn:
agent.invoke(
    {"messages": [{"role": "user", "content": "Now update the tests too."}]},
    config=config,
)

InMemorySaver 的作用仅限于程序运行期间。若需在重启后仍保持数据持久化，应改用基于数据库的检查点机制。

一个代理中的所有组件

这些组件组合在一起便构成一个代理：包括核心循环、自定义工具、规划模块、上下文处理功能、辅助子代理、外壳程序以及保存的状态，总计七个部分。

from deepagents import create_deep_agent
from deepagents.backends import LocalShellBackend
from langchain_core.tools import tool
from langgraph.checkpoint.memory import InMemorySaver

# --- A custom tool (Part 2) — token-budgeted, see Part 2 for the full body ---
@tool
def run_tests(path: str = ".") -> str:
    """Run the project's pytest suite and return its (truncated) output."""
    import subprocess
    try:
        result = subprocess.run(["pytest", path], capture_output=True,
                                text=True, check=False, timeout=300)
    except subprocess.TimeoutExpired:
        return "pytest timed out after 300s"
    output = result.stdout + result.stderr
    return output if len(output) <= 20_000 else "[... truncated ...]\n" + output[-20_000:]

# --- A specialized subagent (Part 5) — note the `system_prompt` key ---
code_searcher = {
    "name": "code-searcher",
    "description": "Finds where specific logic lives in the codebase. "
                   "Use for open-ended 'where is X?' questions.",
    "system_prompt": "You navigate codebases using grep and glob, then report a "
                     "concise summary of what you found. You never make edits.",
}

# --- A system prompt that teaches good behavior (Parts 2 & 3) ---
SYSTEM_PROMPT = """You are a careful coding assistant.
Workflow:
1. Plan the task as a to-do list before doing anything.
2. Use your built-in read, grep, and glob tools to explore - never the raw shell
   equivalents like cat or grep.
3. Make focused edits.
4. ALWAYS run the tests after editing, and fix anything that breaks.
5. Delegate broad codebase searches to the code-searcher subagent.
"""

# --- Assemble the agent (Parts 1, 4, 6, 7 handled by the harness) ---
agent = create_deep_agent(
    model="anthropic:claude-sonnet-4-6",                  # Part 1: the loop, model-agnostic
    tools=[run_tests],                                     # Part 2: custom hands
    system_prompt=SYSTEM_PROMPT,                           # Parts 2 & 3: behavior + planning
    subagents=[code_searcher],                             # Part 5: delegation
    backend=LocalShellBackend(root_dir=".", virtual_mode=False),  # Part 6: shell access
    interrupt_on={"execute": True,                         # Part 6: the brakes —
                  "write_file": True, "edit_file": True},  #   approval before anything destructive
    checkpointer=InMemorySaver(),                          # Part 7: memory across turns
)
# Part 4 (context management) and built-in planning come on automatically.

config = {"configurable": {"thread_id": "my-project"}}
result = agent.invoke(
    {"messages": [{"role": "user", "content": "The login tests are failing. Fix them."}]},
    config=config,
)
print(result["messages"][-1].content)

不到一百行代码就能构建出一个具备规划、搜索、编辑、运行测试、将任务分配给辅助工具、在沙箱环境中运作并保持上下文意识的编程代理。它的结构与 Claude Code 的类似：由工具、规划模块、内存管理机制、任务分配功能以及位于模型之外的约束机制共同构成的简单循环结构。