长时程Agent革命：2026年AI从"聊天机器人"到"自主工作者"的终极跨越
作者: 程序员茄子
日期: 2026-05-16
标签: #AI Agent #长时程任务 #LLM #AGI #多智能体协作
关键词: Long-Horizon Tasks, Agentic AI, LLM, Autonomous Agents, Multi-Agent Systems
引言：AI的"成年礼"
2026年，AI领域正在经历一场静默但深刻的革命。
如果你关注AI的发展，你会发现一个有趣的现象：过去三年，大模型的核心竞赛是"参数规模"——从GPT-3的175B到GPT-5的万亿级参数，从Qwen-72B到Qwen3-Max的万亿参数。但2026年，游戏规则变了。
红杉资本在2026年1月的重磅文章中指出："长时程Agent（Long-Horizon Agent）实际上就是通用人工智能（AGI），2026年将是他们的一年。"
LangChain CEO Harrison Chase也明确表态："2026年是长时程Agents元年。"
这不是巧合。当你和ChatGPT聊天下棋时，Devin已经在GitHub上独立接单写代码；当你还在手动测试API时，Klarna的AI助手已经承担了700名全职客服的工作量。
AI已经从"回答问题"进化到"完成任务"，而现在，它正在向"持续工作数周"进化。
这就是长时程Agent（Long-Horizon Tasks）的本质：让AI能够自主规划、执行、调整一个需要数周甚至数月才能完成的复杂任务，而不需要人类每一步都参与。
本文将深入剖析：
- 什么是长时程Agent，为什么它被认为是AGI的必经之路
- 核心技术架构：从记忆管理到自我反思
- 代码实战：用LangGraph构建一个长时程Agent
- 工业界最佳实践：Devin、DeerFlow、Claude Code的架构揭秘
- 性能优化：如何让Agent持续工作数周不"偏航"
- 未来展望：从一人公司到无人公司
第一章 长时程Agent：从"工具"到"同事"的范式转移
1.1 传统AI的三大局限
要理解长时程Agent的革命性，我们首先需要看清传统AI的三大局限：
局限一：短时程思维（Short-Horizon Thinking）
传统的大语言模型（LLM）本质上是"短时程"的。无论是GPT-5还是Claude Opus，它们的训练目标都是"预测下一个Token"或"解决当前问题"。
问题来了：当你问AI"帮我开发一个电商网站"时，它会给你一个完整的方案，但它不会：
- 自动拆解任务（前端、后端、数据库、部署）
- 自主执行数周的开发工作
- 在遇到Bug时自己Debug并继续前进
- 在项目偏离方向时自我纠正
传统AI像一个"天才顾问"，给你最好的建议，但不会动手帮你做。
局限二：无状态记忆（Stateless Memory）
绝大多数AI应用都是"无状态"的：每次对话都是新的开始，AI不记得上周你让它做的事情。
即使用RAG（检索增强生成）或向量数据库做记忆，传统AI的记忆仍然是"片段式"的：
- 它记得你说过"用React开发"，但不记得你为什么选择React
- 它记得你遇到过某个Bug，但不记得你是怎么解决的
- 它记得项目的技术栈，但不记得项目的长期目标演进路径
长时程任务需要的是"连续记忆"：AI需要像人类一样，在执行任务的过程中不断积累经验、学习教训、调整策略。
局限三：无主动目标驱动（No Goal-Driven Autonomy）
传统AI是"被动响应"的：你问它，它回答；你让它做，它执行。它没有"主动性"。
但真实的工作场景是"目标驱动"的：
- 你告诉AI"帮我提升网站的SEO排名"
- AI需要自己制定计划：关键词分析 → 内容优化 → 技术SEO → 外链建设 → 效果追踪
- AI需要自己决定：今天做什么，明天做什么，遇到瓶颈时如何调整策略
- AI需要自己判断：什么时候任务完成了，什么时候需要向你汇报
这才是"Agent"（智能体）的本质：从被动响应进化到主动规划执行。
1.2 长时程Agent的三大核心能力
长时程Agent（Long-Horizon Agent）之所以被认为是AGI的必经之路，是因为它具备了传统AI缺失的三大核心能力：
能力一：任务分解与层次化规划（Hierarchical Task Planning）
长时程Agent能够将"宏大目标"自动拆解为"可执行的小任务"，并构建层次化的执行计划。
举个例子：
你告诉Agent："帮我开发一个类似Twitter的社交平台。"
传统AI会给你一个文档，列出技术选型和架构建议。
长时程Agent会这样做：
总目标：开发Twitter克隆版
│
├── 阶段一：需求分析与设计（3天）
│ ├── 功能清单整理
│ ├── 数据库表结构设计
│ ├── API接口设计
│ └── 前端页面原型
│
├── 阶段二：后端开发（10天）
│ ├── 用户认证模块
│ ├── 推文CRUD接口
│ ├── 关注/粉丝关系
│ ├── 时间线算法
│ └── 通知系统
│
├── 阶段三：前端开发（10天）
│ ├── 登录注册页面
│ ├── 首页时间线
│ ├── 推文详情页
│ ├── 用户个人主页
│ └── 搜索与发现
│
├── 阶段四：测试与优化（5天）
│ ├── 单元测试
│ ├── 集成测试
│ ├── 性能优化
│ └── 安全加固
│
└── 阶段五：部署与上线（2天）
├── Docker容器化
├── CI/CD流水线
├── 云服务部署
└── 域名与HTTPS
关键不在于"拆解"本身，而在于：
- Agent能够根据执行反馈动态调整计划（如果发现某个任务比预期复杂，会自动增加时间或调整优先级）
- Agent能够在执行过程中自主决策（如果某个技术方案遇到困难，会主动尝试替代方案）
- Agent能够并行执行多个子任务（在等待后端API开发时，先开始前端原型设计）
能力二：长期记忆与经验积累（Long-Term Memory & Experience Accumulation）
长时程Agent具备"类人记忆"：它不仅记住对话历史，还能：
- ** episodic memory（情景记忆）**：记住具体发生过的事情（"2026-05-10，我在开发用户认证模块时遇到了JWT过期问题，通过XXX方案解决了"）
- ** semantic memory（语义记忆）**：记住抽象的知识和规则（"JWT的最佳实践是设置双Token机制：access_token短周期 + refresh_token长周期"）
- ** procedural memory（程序记忆）**：记住解决问题的步骤和策略（"遇到跨域问题，先检查CORS配置，再检查Nginx反向代理，最后检查Cookie的SameSite属性"）
技术实现上，这需要：
- 向量数据库 + 图数据库的混合记忆系统：向量数据库存储语义相似的记忆，图数据库存储记忆之间的关联关系
- 记忆压缩与摘要：长时程任务会产生大量记忆，需要定期压缩（类似人类睡眠时的记忆整理）
- 记忆检索与推理：Agent需要能够快速检索相关记忆，并基于记忆进行推理（"上次遇到类似Bug是怎么解决的？"）
能力三：自我反思与纠错（Self-Reflection & Error Correction）
这是长时程Agent最迷人的能力：它能够"自我反思"，在执行任务的过程中不断优化自己。
人类程序员的工作方式：
- 写代码
- 运行测试，发现Bug
- 分析Bug原因
- 修复Bug
- 总结教训，避免再犯
传统AI的工作方式：
- 生成代码
- 如果出错，重新生成
- 重复1-2，直到用户放弃或凑巧成功
长时程Agent的工作方式：
- 生成代码
- 运行测试，发现Bug
- 自我分析：为什么会出现这个Bug？是我的逻辑错误，还是对需求理解有误？
- 自我纠正：修改代码，并记录这次错误（避免再犯）
- 自我优化：如果发现某种模式的错误反复出现，主动学习并改进自己的"编程风格"
这就像AI从"小学生"进化成了"会自我学习的博士生"。
1.3 为什么2026年是长时程Agent的元年？
你可能会问："这些能力听起来很美好，但为什么是2026年？而不是2025年或2027年？"
答案在于技术成熟度的"临界点"：
临界点一：大模型推理能力的突破
2025年底到2026年初，多个大模型在推理能力上取得了突破性进展：
- OpenAI o1/o3系列：引入了"链式推理（Chain of Thought）"和"自我博弈（Self-Play）"，在MATH竞赛题上的准确率达到90%+
- Claude Opus 4：引入了"持久化记忆"功能，能够在多次对话中保持上下文
- Qwen3.5系列：通过"混合专家（MoE）"架构，在保持推理质量的同时大幅降低了成本
这些进展让AI真正具备了"思考能力"，而不只是"生成能力"。
临界点二：多智能体协作框架的成熟
2026年，多个多智能体协作框架走向成熟：
- LangGraph（LangChain）：提供了状态图（StateGraph）抽象，让开发者能够轻松构建多步骤、有条件的Agent工作流
- DeerFlow（字节跳动）：提供了"超级智能体运行时"，支持多Agent并行执行、沙箱隔离、长期记忆
- AutoGen（微软）：提供了"对话式多智能体"框架，让多个Agent能够通过对话协作完成任务
这些框架让"多Agent协作"从学术论文变成了可用的工具。
临界点三：工具调用与API生态的完善
长时程Agent需要能够"操作工具"：读写文件、执行命令、调用API、浏览网页……
2026年，工具调用（Tool Calling）生态已经非常完善：
- MCP（Model Context Protocol）：Anthropic推出的"AI工具调用标准协议"，让AI能够统一调用各种工具
- Function Calling API：OpenAI、Claude、Qwen等都提供了成熟的Function Calling接口
- 开源工具库：从文件操作到数据库查询，从API调用到网页浏览，几乎所有你能想到的工具都有开源实现
这让Agent能够真正"介入现实世界"，而不只是"纸上谈兵"。
第二章 核心技术架构：长时程Agent是如何工作的？
这一章，我们将深入长时程Agent的技术架构。我会用代码实战的方式，带你从零构建一个能够"持续工作数周"的Agent。
2.1 架构概览：五层金字塔模型
一个完整的长时程Agent系统，通常包含五层架构：
┌─────────────────────────────────────────┐
│ 第五层：人机协作界面（HCI） │ ← 人类与Agent交互的界面
├─────────────────────────────────────────┤
│ 第四层：任务规划与执行引擎（Planner） │ ← 把目标拆解为任务，并调度执行
├─────────────────────────────────────────┤
│ 第三层：记忆与知识管理（Memory） │ ← 管理长期记忆、经验、知识库
├─────────────────────────────────────────┤
│ 第二层：推理与决策引擎（Reasoner） │ ← 基于LLM的推理、规划、决策
├─────────────────────────────────────────┤
│ 第一层：工具与行动接口（Tools） │ ← 文件操作、API调用、代码执行
└─────────────────────────────────────────┘
让我们自下而上，逐层剖析。
2.2 第一层：工具与行动接口（Tools）
核心职责：让Agent能够"介入现实世界"——读写文件、执行命令、调用API、浏览网页。
2.2.1 工具调用的基本原理
大语言模型本身只能"生成文本"，它无法直接操作文件或调用API。要让AI能够使用工具，需要通过**Function Calling（函数调用）**机制。
工作流程：
- 开发者定义一组"工具"（Python函数），并描述它们的作用、参数、返回值
- 将工具定义传递给LLM（通过Function Calling API）
- LLM根据任务需求，决定调用哪个工具，并生成调用参数
- 开发者执行工具调用，获得结果
- 将结果返回给LLM，LLM根据结果决定下一步行动
代码示例：定义一个"读取文件"工具
from typing import List, Dict, Any
import os
# 定义工具
def read_file(filepath: str) -> str:
"""读取文件内容
Args:
filepath: 文件路径
Returns:
文件内容（字符串）
"""
with open(filepath, 'r', encoding='utf-8') as f:
return f.read()
# 工具定义（传递给LLM的"工具说明书"）
READ_FILE_TOOL = {
"type": "function",
"function": {
"name": "read_file",
"description": "读取指定文件路径的内容。支持文本文件、代码文件等。",
"parameters": {
"type": "object",
"properties": {
"filepath": {
"type": "string",
"description": "要读取的文件路径（绝对路径或相对路径）"
}
},
"required": ["filepath"]
}
}
}
# 工具注册表
AVAILABLE_TOOLS = {
"read_file": read_file
}
TOOL_DEFINITIONS = [READ_FILE_TOOL]
2.2.2 让LLM调用工具（OpenAI兼容API）
from openai import OpenAI
client = OpenAI(
api_key="your-api-key",
base_url="https://api.openai.com/v1" # 或国产模型的API地址
)
def agent_think_and_act(user_message: str) -> str:
"""让Agent思考并可能调用工具"""
# 第一轮：让LLM思考是否需要调用工具
response = client.chat.completions.create(
model="gpt-4o",
messages=[
{"role": "system", "content": "你是一个能够使用工具的AI助手。"},
{"role": "user", "content": user_message}
],
tools=TOOL_DEFINITIONS, # 传入工具定义
tool_choice="auto" # 让LLM自己决定是否调用工具
)
message = response.choices[0].message
# 检查LLM是否要求调用工具
if message.tool_calls:
tool_results = []
for tool_call in message.tool_calls:
function_name = tool_call.function.name
function_args = json.loads(tool_call.function.arguments)
# 执行工具调用
if function_name in AVAILABLE_TOOLS:
result = AVAILABLE_TOOLS[function_name](**function_args)
tool_results.append({
"tool_call_id": tool_call.id,
"role": "tool",
"name": function_name,
"content": str(result)
})
# 第二轮：将工具结果返回给LLM，让它生成最终回复
final_response = client.chat.completions.create(
model="gpt-4o",
messages=[
{"role": "system", "content": "你是一个能够使用工具的AI助手。"},
{"role": "user", "content": user_message},
message, # LLM要求调用工具的那个message
*tool_results # 工具调用的结果
]
)
return final_response.choices[0].message.content
else:
# LLM没有调用工具，直接返回回复
return message.content
这就是Agent能够"行动"的基本原理。通过Function Calling，LLM能够"看到"工具的定义，决定何时调用哪个工具，并根据工具返回的结果调整自己的行动。
2.2.3 常用工具分类
一个长时程Agent通常需要以下类型的工具：
| 工具类别 | 具体工具 | 用途 |
|---|---|---|
| 文件操作 | read_file, write_file, list_directory | 读写代码、配置文件 |
| 命令执行 | execute_shell, run_python | 执行脚本、运行测试 |
| API调用 | http_request, api_call | 调用第三方服务 |
| 网页浏览 | browse_web, search_web | 搜索资料、查阅文档 |
| 数据库操作 | query_db, insert_db | 存储和检索数据 |
| 代码执行 | execute_code, run_tests | 运行代码、验证结果 |
实战建议：不要一次性实现所有工具。先从最核心的3-5个工具开始（文件读写 + Shell执行 + API调用），逐步扩展。
2.3 第二层：推理与决策引擎（Reasoner）
核心职责：基于LLM，让Agent具备"思考能力"——分析问题、制定计划、做出决策。
2.3.1 推理模式：ReAct vs. Plan-and-Execute
目前主流的Agent推理模式有两种：
模式一：ReAct（Reasoning + Acting）
核心思想：让Agent交替进行"推理（Reasoning）"和"行动（Acting）"。
用户输入 → Agent思考 → Agent行动（调用工具） → 观察结果 → Agent思考 → Agent行动 → ...
代码示例（简化版）：
def react_agent(user_task: str, max_steps: int = 50):
"""ReAct模式的Agent"""
messages = [
{"role": "system", "content": """你是一个能够使用工具的AI助手。
你的任务是帮助用户完成复杂任务。请按以下步骤思考：
1. Thought（思考）：分析当前状态，决定下一步做什么
2. Action（行动）：调用合适的工具
3. Observation（观察）：观察工具执行结果
4. 重复1-3，直到任务完成
5. Final Answer（最终答案）：给出任务结果
"""},
{"role": "user", "content": user_task}
]
for step in range(max_steps):
# 让LLM思考并行动
response = client.chat.completions.create(
model="gpt-4o",
messages=messages,
tools=TOOL_DEFINITIONS,
tool_choice="auto"
)
message = response.choices[0].message
messages.append(message)
# 如果LLM调用了工具
if message.tool_calls:
for tool_call in message.tool_calls:
function_name = tool_call.function.name
function_args = json.loads(tool_call.function.arguments)
# 执行工具
result = AVAILABLE_TOOLS[function_name](**function_args)
# 将结果添加到对话
messages.append({
"tool_call_id": tool_call.id,
"role": "tool",
"name": function_name,
"content": str(result)
})
# 如果LLM给出了最终答案（没有工具调用）
else:
print(f"✅ 任务完成！")
print(message.content)
break
else:
print(f"⚠️ 达到最大步数（{max_steps}），任务可能未完成。")
ReAct的优缺点：
- ✅ 优点：灵活，能够动态调整行动
- ❌ 缺点：缺乏长期规划，容易"短视"（适合短时程任务）
模式二：Plan-and-Execute（规划-执行）
核心思想：先制定完整计划，再逐步执行。
用户输入 → 制定计划（Plan） → 执行步骤1 → 执行步骤2 → ... → 任务完成
代码示例（简化版）：
def plan_and_execute_agent(user_task: str):
"""Plan-and-Execute模式的Agent"""
# 第一阶段：让LLM制定计划
planning_prompt = f"""
用户任务：{user_task}
请为这个任务制定一个详细的执行计划。
要求：
1. 将任务分解为3-10个步骤
2. 每个步骤明确、可执行
3. 考虑依赖关系（哪些步骤需要等待其他步骤完成）
输出格式（JSON）：
{{
"plan": [
{{"step": 1, "description": "步骤1的描述", "depends_on": []}},
{{"step": 2, "description": "步骤2的描述", "depends_on": [1]}},
...
]
}}
"""
response = client.chat.completions.create(
model="gpt-4o",
messages=[{"role": "user", "content": planning_prompt}],
response_format={"type": "json_object"} # 强制输出JSON
)
plan_data = json.loads(response.choices[0].message.content)
plan = plan_data["plan"]
print(f"📋 执行计划：")
for step_info in plan:
print(f" 步骤{step_info['step']}: {step_info['description']}")
# 第二阶段：按计划执行
results = {{}} # 存储每个步骤的结果
for step_info in plan:
step_num = step_info['step']
description = step_info['description']
depends_on = step_info['depends_on']
print(f"
🔧 执行步骤{step_num}：{description}")
# 检查依赖
if depends_on:
dependencies = "
".join([f"步骤{dep}: {results[dep]}" for dep in depends_on])
context = f"依赖的步骤结果：
{dependencies}"
else:
context = ""
# 执行步骤
execution_prompt = f"""
任务：{description}
{context}
请执行这个步骤。如果需要使用工具，请调用相应的工具。
"""
response = client.chat.completions.create(
model="gpt-4o",
messages=[{"role": "user", "content": execution_prompt}],
tools=TOOL_DEFINITIONS,
tool_choice="auto"
)
# 处理工具调用（简化，完整代码需要处理多轮工具调用）
result = response.choices[0].message.content
results[step_num] = result
print(f" 完成：{result[:100]}...")
print(f"
✅ 所有步骤执行完成！")
Plan-and-Execute的优缺点：
- ✅ 优点：有全局视野，适合长时程任务
- ❌ 缺点：计划可能不完美，执行过程中需要动态调整
2.3.2 最佳实践：ReAct + Plan-and-Execute 混合模式
工业界的最佳实践是混合模式：
- 高层规划：用Plan-and-Execute制定宏观计划（阶段 → 大步骤）
- 低层执行：用ReAct执行每个大步骤（灵活应对细节问题）
这就是Devin、DeerFlow等顶级Agent使用的架构。
2.4 第三层：记忆与知识管理（Memory）
核心职责：让Agent具备"长期记忆"，能够在数周的任务执行过程中积累经验、学习教训。
2.4.1 记忆系统的三大挑战
挑战一：上下文窗口限制
即使最先进的大模型（如Claude Opus 4、GPT-5），其上下文窗口也是有限的（通常200K-1M tokens）。
问题：如果一个任务需要执行数周，产生数百万tokens的对话历史、代码、日志，远超上下文窗口。
解决方案：记忆压缩 + 检索增强生成（RAG）
挑战二：记忆的"信噪比"
不是所有历史信息都值得记住。如果Agent记住了每一个细节（包括无用的），会导致：
- 检索效率低下
- 重要信息被淹没
- 推理速度变慢
解决方案：重要性评分 + 定期遗忘
挑战三：记忆的"时效性"
某些记忆会"过期"。比如：
- "项目使用Python 3.10" → 如果项目升级到Python 3.12，这条记忆就过时了
- "某个API的调用方式" → 如果API升级了，旧记忆就误导Agent
解决方案：时间戳 + 版本管理
2.4.2 记忆系统的技术实现
一个完整的记忆系统通常包含以下组件：
┌──────────────────────────────────────┐
│ 记忆管理器（MemoryManager） │
├──────────────────────────────────────┤
│ • 记忆存储（向量数据库 + 图数据库） │
│ • 记忆检索（语义搜索 + 关联查询） │
│ • 记忆压缩（LLM摘要 + 关键信息提取） │
│ • 记忆遗忘（基于时间 + 重要性） │
└──────────────────────────────────────┘
代码示例：使用LangChain的记忆管理系统
from langchain.memory import ConversationBufferMemory, ConversationSummaryMemory
from langchain.vectorstores import Chroma
from langchain.embeddings import OpenAIEmbeddings
from langchain.schema import Document
class LongHorizonMemory:
"""长时程Agent的记忆管理系统"""
def __init__(self, agent_id: str):
self.agent_id = agent_id
# 1. 短期记忆（最近N轮对话，完整保存）
self.short_term_memory = ConversationBufferMemory(
memory_key="chat_history",
return_messages=True,
max_token_limit=100000 # 最近10万tokens完整保存
)
# 2. 长期记忆（向量数据库，语义检索）
self.embeddings = OpenAIEmbeddings()
self.long_term_memory = Chroma(
collection_name=f"agent_{agent_id}_long_term",
embedding_function=self.embeddings
)
# 3. 记忆压缩器（将短期记忆压缩后存入长期记忆）
self.memory_compressor = ConversationSummaryMemory(
llm=client,
max_token_limit=50000
)
def add(self, interaction: dict):
"""添加一个交互记录（用户输入 + Agent回复 + 工具调用结果）"""
# 1. 存入短期记忆
self.short_term_memory.save_context(
{"input": interaction["user_input"]},
{"output": interaction["agent_output"]}
)
# 2. 存入长期记忆（向量化）
doc = Document(
page_content=f"用户：{interaction['user_input']}
Agent：{interaction['agent_output']}",
metadata={
"timestamp": time.time(),
"importance": self._calculate_importance(interaction),
"tags": self._extract_tags(interaction)
}
)
self.long_term_memory.add_documents([doc])
# 3. 如果短期记忆过大，压缩后存入长期记忆
if self.short_term_memory.buffer_length() > 50: # 超过50轮对话
self._compress_and_store()
def retrieve(self, query: str, k: int = 5) -> List[Document]:
"""检索相关记忆"""
# 1. 从长期记忆中语义搜索
docs = self.long_term_memory.similarity_search(query, k=k)
# 2. 从短期记忆中获取最近对话
recent = self.short_term_memory.load_memory_variables({})
return docs, recent
def _compress_and_store(self):
"""压缩短期记忆，存入长期记忆"""
# 使用LLM将最近N轮对话压缩成摘要
summary = self.memory_compressor.predict_new_summary(
messages=self.short_term_memory.chat_memory.messages,
existing_summary=""
)
# 将摘要存入长期记忆
doc = Document(
page_content=summary,
metadata={
"timestamp": time.time(),
"type": "summary",
"importance": 0.8 # 摘要通常比较重要
}
)
self.long_term_memory.add_documents([doc])
# 清空短期记忆（保留最近5轮）
# ...（省略具体实现）
def _calculate_importance(self, interaction: dict) -> float:
"""计算记忆的重要性（0-1）"""
# 简单启发式：如果交互包含"错误"、"Bug"、"成功"等关键词，重要性更高
keywords_high = ["error", "bug", "failed", "success", "completed"]
keywords_low = ["hello", "thanks", "ok"]
text = interaction["user_input"] + " " + interaction["agent_output"]
if any(kw in text.lower() for kw in keywords_high):
return 0.8
elif any(kw in text.lower() for kw in keywords_low):
return 0.2
else:
return 0.5
def _extract_tags(self, interaction: dict) -> List[str]:
"""提取记忆的标签（用于分类检索）"""
# 使用LLM提取关键词
prompt = f"""
请从以下对话中提取3-5个关键词（技术名词、工具名、概念名等）：
对话：{interaction['user_input']} {interaction['agent_output']}
输出格式（JSON）：
{{"tags": ["tag1", "tag2", "tag3"]}}
"""
response = client.chat.completions.create(
model="gpt-4o-mini", # 用便宜的模型
messages=[{"role": "user", "content": prompt}],
response_format={"type": "json_object"}
)
return json.loads(response.choices[0].message.content)["tags"]
这个记忆系统的亮点：
- 分层存储：短期记忆（完整）+ 长期记忆（压缩）
- 语义检索：用向量数据库实现"模糊回忆"
- 重要性评分：重要记忆优先级更高
- 自动压缩：避免记忆无限增长
2.5 第四层：任务规划与执行引擎（Planner）
核心职责：将用户的"宏大目标"拆解为"可执行计划"，并调度执行。
2.5.1 任务分解的核心算法
任务分解（Task Decomposition）是长时程Agent最核心的能力之一。主流算法有三种：
算法一：递归分解（Recursive Decomposition）
思想：将任务递归地拆解为子任务，直到子任务可以直接执行为止。
任务：开发一个电商网站
├── 子任务1：后端开发
│ ├── 子任务1.1：数据库设计
│ ├── 子任务1.2：API开发
│ └── 子任务1.3：身份验证
├── 子任务2：前端开发
│ ├── 子任务2.1：UI设计
│ └── 子任务2.2：交互开发
└── 子任务3：部署
├── 子任务3.1：Docker容器化
└── 子任务3.2：云服务部署
代码示例：
def recursive_decompose(task: str, max_depth: int = 5) -> dict:
"""递归分解任务"""
if max_depth == 0:
return {"task": task, "executable": True} # 达到最大深度，直接执行
# 让LLM判断这个任务是否可以直接执行
check_prompt = f"""
任务：{task}
这个任务是否可以直接执行（不需要进一步分解）？
如果可以直接执行，回答：{{"executable": true}}
如果需要分解，回答：{{"executable": false, "subtasks": ["子任务1", "子任务2", ...]}}
要求：如果需要分解，子任务数量3-7个。
"""
response = client.chat.completions.create(
model="gpt-4o",
messages=[{"role": "user", "content": check_prompt}],
response_format={"type": "json_object"}
)
result = json.loads(response.choices[0].message.content)
if result["executable"]:
return {"task": task, "executable": True}
else:
# 递归分解子任务
subtasks = []
for subtask in result["subtasks"]:
subtasks.append(recursive_decompose(subtask, max_depth - 1))
return {"task": task, "executable": False, "subtasks": subtasks}
算法二：基于依赖图的分解（Dependency Graph Based Decomposition）
思想：不仅分解任务,还构建任务之间的依赖关系（哪些任务需要等待其他任务完成）。
def decompose_with_dependencies(task: str) -> dict:
"""基于依赖图的任务分解"""
prompt = f"""
任务：{task}
请将这个任务分解为多个子任务，并明确子任务之间的依赖关系。
输出格式（JSON）：
{{
"tasks": [
{{"id": 1, "description": "子任务1", "depends_on": []}},
{{"id": 2, "description": "子任务2", "depends_on": [1]}},
{{"id": 3, "description": "子任务3", "depends_on": [1, 2]}},
...
]
}}
要求：
1. 子任务数量5-15个
2. 明确标注依赖关系（depends_on字段）
3. 考虑并行性（没有依赖关系的任务可以并行执行）
"""
response = client.chat.completions.create(
model="gpt-4o",
messages=[{"role": "user", "content": prompt}],
response_format={"type": "json_object"}
)
return json.loads(response.choices[0].message.content)
算法三：层次化规划（Hierarchical Planning）
思想：将任务分解为"阶段 → 步骤 → 具体操作"三个层次。
这是DeerFlow、Devin等顶级Agent使用的算法。
def hierarchical_plan(task: str) -> dict:
"""层次化规划"""
# 第一层：规划阶段（Phases）
phase_prompt = f"""
任务：{task}
请将这个任务分解为3-7个阶段（Phases）。每个阶段是一个宏观的工作单元。
输出格式（JSON）：
{{
"phases": [
{{"id": "P1", "name": "需求分析", "description": "..."}},
{{"id": "P2", "name": "系统设计", "description": "..."}},
...
]
}}
"""
phase_response = client.chat.completions.create(
model="gpt-4o",
messages=[{"role": "user", "content": phase_prompt}],
response_format={"type": "json_object"}
)
phases = json.loads(phase_response.choices[0].message.content)["phases"]
# 第二层：为每个阶段规划步骤（Steps）
for phase in phases:
step_prompt = f"""
阶段：{phase['name']}
描述：{phase['description']}
请将这个阶段分解为3-7个具体步骤（Steps）。
输出格式（JSON）：
{{
"steps": [
{{"id": "S1", "description": "步骤1", "depends_on": []}},
{{"id": "S2", "description": "步骤2", "depends_on": ["S1"]}},
...
]
}}
"""
step_response = client.chat.completions.create(
model="gpt-4o",
messages=[{"role": "user", "content": step_prompt}],
response_format={"type": "json_object"}
)
phase["steps"] = json.loads(step_response.choices[0].message.content)["steps"]
# 第三层：为每个步骤规划具体操作（Actions）
for step in phase["steps"]:
action_prompt = f"""
步骤：{step['description']}
请将这个步骤分解为具体的操作（Actions）。每个操作应该是可以直接执行的（调用工具或生成代码）。
输出格式（JSON）：
{{
"actions": [
{{"id": "A1", "description": "操作1", "tool": "write_file", "args": {{...}}}},
{{"id": "A2", "description": "操作2", "tool": "execute_command", "args": {{...}}}},
...
]
}}
"""
action_response = client.chat.completions.create(
model="gpt-4o-mini", # 用便宜的模型
messages=[{"role": "user", "content": action_prompt}],
response_format={"type": "json_object"}
)
step["actions"] = json.loads(action_response.choices[0].message.content)["actions"]
return {"phases": phases}
2.5.2 任务执行的调度策略
有了任务计划后，还需要一个**调度器（Scheduler）**来决定：
- 哪个任务先执行？
- 哪些任务可以并行执行？
- 如果某个任务失败，如何处理？
代码示例：一个简单的基于依赖关系的调度器
from collections import defaultdict, deque
class TaskScheduler:
"""任务调度器"""
def __init__(self, tasks: List[dict]):
self.tasks = {{task["id"]: task for task in tasks}}
self.dependencies = defaultdict(list) # task_id -> [依赖的task_id列表]
self.reverse_deps = defaultdict(list) # task_id -> [依赖于它的task_id列表]
self.status = {{}} # task_id -> "pending" | "running" | "completed" | "failed"
# 构建依赖图
for task in tasks:
task_id = task["id"]
self.status[task_id] = "pending"
for dep in task.get("depends_on", []):
self.dependencies[task_id].append(dep)
self.reverse_deps[dep].append(task_id)
def get_ready_tasks(self) -> List[str]:
"""获取所有'可以继续执行'的任务（依赖已完成，且状态为pending）"""
ready = []
for task_id, deps in self.dependencies.items():
if self.status[task_id] == "pending":
if all(self.status[dep] == "completed" for dep in deps):
ready.append(task_id)
return ready
def mark_completed(self, task_id: str):
"""标记任务完成"""
self.status[task_id] = "completed"
def mark_failed(self, task_id: str):
"""标记任务失败"""
self.status[task_id] = "failed"
def execute_all(self):
"""执行所有任务（简化版，实际应该用异步执行）"""
while True:
ready_tasks = self.get_ready_tasks()
if not ready_tasks:
break # 没有可执行的任务了
for task_id in ready_tasks:
print(f"▶️ 执行任务 {task_id}: {self.tasks[task_id]['description']}")
self.status[task_id] = "running"
try:
# 执行任务（这里简化为打印）
# 实际应该调用Agent执行
result = self._execute_task(self.tasks[task_id])
print(f" ✅ 完成：{result[:50]}...")
self.mark_completed(task_id)
except Exception as e:
print(f" ❌ 失败：{e}")
self.mark_failed(task_id)
# 失败策略：终止所有依赖于它的任务
self._propagate_failure(task_id)
break
# 打印执行结果
completed = [tid for tid, status in self.status.items() if status == "completed"]
failed = [tid for tid, status in self.status.items() if status == "failed"]
print(f"
📊 执行结果：")
print(f" 完成：{len(completed)}个任务")
print(f" 失败：{len(failed)}个任务")
def _execute_task(self, task: dict) -> str:
"""执行单个任务（简化）"""
# 实际应该调用Agent的推理+行动
return f"任务'{task['description']}'执行完成"
def _propagate_failure(self, failed_task_id: str):
"""传播失败：将依赖于failed_task_id的所有任务标记为failed"""
for dependent in self.reverse_deps[failed_task_id]:
self.status[dependent] = "failed"
self._propagate_failure(dependent) # 递归传播
2.6 第五层：人机协作界面（HCI）
核心职责：让人类能够方便地给Agent分配任务、查看进度、介入调整。
2.6.1 长时程Agent的HCI设计原则
设计长时程Agent的人机界面，需要遵循以下原则：
原则一：进度可见性（Progress Visibility）
人类需要清楚地知道Agent在做什么、做到了哪一步、预计还需要多长时间。
实现方式：
- 实时进度条（每个阶段/步骤的完成百分比）
- 任务树可视化（用树状图展示任务分解结构，已完成/进行中/待开始的用不同颜色标注）
- 时间线视图（类似Gantt图，展示每个任务的开始时间、预计结束时间）
原则二：介入点设计（Intervention Points）
Agent不是完全自主的，人类需要在关键时刻能够介入：
- 审批点：Agent完成一个重要阶段后，需要人类审批才能继续（比如"数据库设计完成，是否继续开发API？"）
- 决策点：Agent遇到多个可选方案时，让人类选择（比如"有两种技术方案，方案A更简单，方案B性能更好，你选哪个？"）
- 紧急停止：人类随时能够暂停/终止Agent的执行
原则三：反馈闭环（Feedback Loop）
人类不仅要"监控"Agent，还要能够"反馈"——指出错误、提出改进建议、调整方向。
实现方式：
- 每个阶段结束后，Agent自动生成"阶段总结报告"，人类可以评论
- Agent定期（比如每天）发送"工作日志"给人类
- 人类可以随时给Agent发送消息（类似给人类同事发微信）
2.6.2 代码实战：一个简单的HCI实现
import streamlit as st # 用Streamlit快速构建Web界面
from datetime import datetime
class AgentHCI:
"""Agent的人机协作界面"""
def __init__(self, agent):
self.agent = agent
def render_dashboard(self):
"""渲染Agent的仪表盘"""
st.title("🤖 长时程Agent工作台")
# 1. 任务总览
st.header("📋 任务总览")
task = self.agent.current_task
st.write(f"**当前任务**：{task['description']}")
st.write(f"**开始时间**：{task['start_time']}")
st.write(f"**预计完成时间**：{task['estimated_end_time']}")
# 2. 进度条
st.header("📊 执行进度")
progress = self.agent.get_progress()
st.progress(progress / 100)
st.write(f"**总体进度**：{progress}%")
# 3. 任务树可视化
st.header("🌲 任务分解树")
task_tree = self.agent.get_task_tree()
self._render_task_tree(task_tree)
# 4. 实时日志
st.header("📝 实时日志")
logs = self.agent.get_recent_logs(limit=20)
for log in logs:
timestamp = log["timestamp"].strftime("%H:%M:%S")
level = log["level"] # "INFO" | "WARNING" | "ERROR"
message = log["message"]
if level == "INFO":
st.text(f"[{timestamp}] ℹ️ {message}")
elif level == "WARNING":
st.text(f"[{timestamp}] ⚠️ {message}")
elif level == "ERROR":
st.text(f"[{timestamp}] ❌ {message}")
# 5. 介入控制
st.header("🎮 介入控制")
col1, col2, col3 = st.columns(3)
with col1:
if st.button("⏸️ 暂停"):
self.agent.pause()
st.success("Agent已暂停")
with col2:
if st.button("▶️ 继续"):
self.agent.resume()
st.success("Agent已继续")
with col3:
if st.button("⏹️ 终止"):
self.agent.terminate()
st.warning("Agent已终止")
# 6. 发送消息给Agent
st.header("💬 与Agent对话")
user_message = st.text_input("给Agent发消息...")
if st.button("发送"):
response = self.agent.receive_message(user_message)
st.write(f"**Agent回复**：{response}")
def _render_task_tree(self, tree: dict, level: int = 0):
"""递归渲染任务树"""
indent = " " * level
for task in tree["tasks"]:
task_id = task["id"]
description = task["description"]
status = task["status"] # "pending" | "running" | "completed" | "failed"
# 根据状态选择图标
if status == "completed":
icon = "✅"
elif status == "running":
icon = "▶️"
elif status == "failed":
icon = "❌"
else:
icon = "⏳"
st.text(f"{indent}{icon} {task_id}: {description}")
# 递归渲染子任务
if "subtasks" in task:
self._render_task_tree({"tasks": task["subtasks"]}, level + 1)
这个HCI界面的功能：
- 实时查看Agent的执行进度
- 可视化任务分解结构
- 查看实时日志
- 暂停/继续/终止Agent
- 与Agent对话（发送消息、接收回复）
第三章 代码实战：用LangGraph构建长时程Agent
理论讲完了,现在让我们动手实战！这一章，我将用**LangGraph（LangChain的图形化Agent框架）**构建一个能够"持续工作"的长时程Agent。
3.1 为什么选择LangGraph？
LangGraph是2026年最流行的Agent开发框架之一，核心优势是：
- 状态图（StateGraph）抽象：将Agent的执行流程建模为"状态图"，每个节点是一个处理步骤，边是状态转移条件
- 循环与条件分支：支持复杂的循环（while、for）和条件分支（if-else）
- 持久化状态：Agent的状态可以持久化到数据库，即使重启也能继续执行
- 人工介入点：可以在流程中设置"人工审批"节点
安装LangGraph：
pip install langgraph langchain-openai
3.2 实战项目：构建一个"自动写技术博客"的Agent
任务描述：让用户给定一个主题，Agent自动：
- 搜索相关资料
- 撰写一篇深度技术博客（5000+字）
- 生成配图（如果有必要）
- 发布到博客平台
这个任务需要数小时甚至数天（如果需要深入研究），是典型的长时程任务。
3.2.1 定义状态（State）
首先，定义Agent的"状态"（在整个执行过程中传递的数据）：
from typing import TypedDict, List, Optional
from langchain_core.messages import HumanMessage, AIMessage
class BlogWritingState(TypedDict):
"""写博客任务的状态"""
# 用户输入
topic: str # 博客主题
target_length: int # 目标字数（默认5000）
# 研究阶段
search_results: List[dict] # 搜索结果
key_points: List[str] # 提取的关键要点
# 写作阶段
outline: dict # 文章大纲
draft: str # 草稿
final_article: str # 最终文章
# 发布阶段
published_url: Optional[str] # 发布后的URL
# 控制流
current_step: str # 当前步骤
error_message: Optional[str] # 错误信息
messages: List[HumanMessage | AIMessage] # 对话历史
3.2.2 定义处理节点（Nodes）
然后，定义每个处理步骤（节点）：
from langchain_openai import ChatOpenAI
from langgraph.graph import StateGraph, END
# 初始化LLM
llm = ChatOpenAI(model="gpt-4o")
# 节点1：搜索资料
def search_research(state: BlogWritingState) -> BlogWritingState:
"""搜索相关资料"""
topic = state["topic"]
print(f"🔍 正在搜索资料：{topic}")
# 调用搜索工具（这里简化为模拟）
# 实际应该调用SerpAPI、Bing Search等
search_results = [
{"title": "文章1", "url": "https://example.com/1", "snippet": "..."},
{"title": "文章2", "url": "https://example.com/2", "snippet": "..."},
# ...
]
state["search_results"] = search_results
state["current_step"] = "research_completed"
return state
# 节点2：提取关键要点
def extract_key_points(state: BlogWritingState) -> BlogWritingState:
"""从搜索结果中提取关键要点"""
search_results = state["search_results"]
print(f"📋 正在提取关键要点...")
# 让LLM提取要点
prompt = f"""
请从以下搜索结果中提取5-10个关键要点（用于写博客）：
搜索结果：
{search_results}
输出格式（JSON）：
{{"key_points": ["要点1", "要点2", ...]}}
"""
response = llm.invoke([HumanMessage(content=prompt)])
key_points = json.loads(response.content)["key_points"]
state["key_points"] = key_points
state["current_step"] = "key_points_extracted"
return state
# 节点3：制定文章大纲
def create_outline(state: BlogWritingState) -> BlogWritingState:
"""制定文章大纲"""
topic = state["topic"]
key_points = state["key_points"]
target_length = state["target_length"]
print(f"📝 正在制定文章大纲...")
prompt = f"""
博客主题：{topic}
关键要点：{key_points}
目标字数：{target_length}字
请为这篇博客制定一个详细的大纲。
要求：
1. 包含引言、3-7个章节、总结
2. 每个章节有明确的标题和子标题
3. 预计每章节字数均匀分布
输出格式（JSON）：
{{
"outline": [
{{"chapter": 1, "title": "引言", "word_count": 500}},
{{"chapter": 2, "title": "第一章标题", "sections": ["小节1.1", "小节1.2"], "word_count": 1000}},
...
]
}}
"""
response = llm.invoke([HumanMessage(content=prompt)])
outline = json.loads(response.content)["outline"]
state["outline"] = outline
state["current_step"] = "outline_created"
return state
# 节点4：撰写草稿
def write_draft(state: BlogWritingState) -> BlogWritingState:
"""根据大纲撰写草稿"""
topic = state["topic"]
outline = state["outline"]
key_points = state["key_points"]
print(f"✍️ 正在撰写草稿...")
# 逐章节撰写（这里简化为一次生成）
prompt = f"""
博客主题：{topic}
文章大纲：{outline}
关键要点：{key_points}
请撰写一篇完整的技术博客。
要求：
1. 严格按照大纲结构
2. 深度技术剖析，有代码示例
3. 语言风格：程序员视角，实用主义
4. 字数：{state['target_length']}字左右
"""
response = llm.invoke([HumanMessage(content=prompt)])
draft = response.content
state["draft"] = draft
state["current_step"] = "draft_completed"
return state
# 节点5：审校与修改
def review_and_revise(state: BlogWritingState) -> BlogWritingState:
"""审校文章，必要时修改"""
draft = state["draft"]
print(f"🔍 正在审校文章...")
prompt = f"""
请审校以下文章草稿：
{draft}
审校要点：
1. 技术准确性（代码示例是否正确？）
2. 逻辑连贯性（章节之间是否流畅？）
3. 深度与广度（是否达到预期质量？）
如果发现问题，请直接修改并提供最终版本。
如果文章质量良好，直接返回原文。
输出格式：
{{"needs_revision": true/false, "final_article": "..."}}
"""
response = llm.invoke([HumanMessage(content=prompt)])
result = json.loads(response.content)
state["final_article"] = result["final_article"]
state["current_step"] = "review_completed"
return state
# 节点6：发布文章
def publish_article(state: BlogWritingState) -> BlogWritingState:
"""发布文章到博客平台"""
final_article = state["final_article"]
topic = state["topic"]
print(f"🚀 正在发布文章...")
# 调用发布API（这里简化为模拟）
# 实际应该调用WordPress API、Medium API等
published_url = "https://blog.example.com/posts/2026-05-16-long-horizon-agent"
state["published_url"] = published_url
state["current_step"] = "published"
print(f"✅ 文章已发布：{published_url}")
return state
3.2.3 构建状态图（Graph）
定义好节点后，用LangGraph构建状态图：
def build_blog_writing_graph():
"""构建写博客的Agent状态图"""
# 创建状态图
graph = StateGraph(BlogWritingState)
# 添加节点
graph.add_node("search", search_research)
graph.add_node("extract", extract_key_points)
graph.add_node("outline", create_outline)
graph.add_node("write", write_draft)
graph.add_node("review", review_and_revise)
graph.add_node("publish", publish_article)
# 添加边（定义执行顺序）
graph.add_edge("search", "extract")
graph.add_edge("extract", "outline")
graph.add_edge("outline", "write")
graph.add_edge("write", "review")
graph.add_edge("review", "publish")
graph.add_edge("publish", END)
# 设置入口点
graph.set_entry_point("search")
# 编译图
return graph.compile()
# 使用Agent
def run_blog_writing_agent(topic: str, target_length: int = 5000):
"""运行写博客的Agent"""
# 构建图
graph = build_blog_writing_graph()
# 初始状态
initial_state = BlogWritingState(
topic=topic,
target_length=target_length,
search_results=[],
key_points=[],
outline={},
draft="",
final_article="",
published_url=None,
current_step="start",
error_message=None,
messages=[]
)
# 执行图
print(f"🤖 开始执行写博客任务：{topic}
")
final_state = graph.invoke(initial_state)
print(f"
🎉 任务完成！")
print(f"文章已发布：{final_state['published_url']}")
return final_state
# 运行Agent
if __name__ == "__main__":
run_blog_writing_agent(
topic="长时程Agent技术深度剖析",
target_length=8000
)
3.3 增强版：添加人工审批节点
在实际应用中，我们可能希望在"发布文章"之前，让人类审批一下。
修改状态图，添加人工审批：
from langgraph.graph import StateGraph, END
from langgraph.checkpoint.sqlite import SqliteSaver
def build_blog_writing_graph_with_approval():
"""构建带人工审批的写博客Agent"""
graph = StateGraph(BlogWritingState)
# 添加节点
graph.add_node("search", search_research)
graph.add_node("extract", extract_key_points)
graph.add_node("outline", create_outline)
graph.add_node("write", write_draft)
graph.add_node("review", review_and_revise)
graph.add_node("approve", human_approval) # 新增：人工审批节点
graph.add_node("publish", publish_article)
# 添加边
graph.add_edge("search", "extract")
graph.add_edge("extract", "outline")
graph.add_edge("outline", "write")
graph.add_edge("write", "review")
graph.add_edge("review", "approve") # 审校后进入审批
# 条件边：根据审批结果决定是发布还是重新撰写
graph.add_conditional_edges(
"approve",
should_publish, # 这个函数检查审批结果
{
"publish": "publish",
"rewrite": "write" # 如果审批不通过，重新撰写
}
)
graph.add_edge("publish", END)
graph.set_entry_point("search")
# 使用SQLite持久化状态（支持中断恢复）
with SqliteSaver.from_conn_string("agent_state.db") as checkpointer:
return graph.compile(checkpointer=checkpointer)
def human_approval(state: BlogWritingState) -> BlogWritingState:
"""人工审批节点"""
final_article = state["final_article"]
print(f"📝 文章草稿已完成，请审批：")
print(f"{'='*50}")
print(final_article[:500] + "...") # 只显示前500字
print(f"{'='*50}")
# 在实际应用中，这里应该暂停并等待人类输入
# LangGraph提供了interrupt功能
approval = input("是否批准发布？(yes/no): ")
if approval.lower() == "yes":
state["current_step"] = "approved"
else:
state["current_step"] = "rejected"
state["error_message"] = input("拒绝原因（可选）：")
return state
def should_publish(state: BlogWritingState) -> str:
"""决定是否发布"""
if state["current_step"] == "approved":
return "publish"
else:
return "rewrite"
这个增强版Agent的特性：
- 在发布前需要人类审批
- 如果审批不通过，可以重新撰写
- 状态持久化到数据库，即使中断也能恢复
第四章 工业界最佳实践：顶级Agent的架构揭秘
理论和完善的代码实战讲完了，现在让我们看看工业界的顶级Agent是如何设计的。
4.1 Devin：自主软件工程师
Devin（Cognition AI开发）是2026年最著名的AI软件工程师，能够独立接单写代码。
架构特点
- 沙箱执行环境：Devin运行在独立的Docker容器中，可以安全地执行任意代码
- 长期记忆：Devin记住每个项目的上下文（技术栈、代码风格、团队规范）
- 自主调试：遇到Bug时，Devin能够自己Google错误、阅读文档、尝试修复
- 人类协作：在关键决策点（比如选择架构方案），Devin会主动询问人类
核心技术
- 工具集：文件系统、Shell、浏览器、GitHub API、Stack Overflow搜索
- 推理模式：ReAct + Plan-and-Execute混合
- 记忆系统：向量数据库（Chroma） + 结构化数据库（PostgreSQL）
4.2 DeerFlow：字节跳动的超级智能体运行时
DeerFlow（字节跳动开源）是一个"超级智能体运行时"，支持多Agent并行执行。
架构特点
- 基于LangGraph：使用LangGraph构建多Agent工作流
- Docker沙箱：每个Agent运行在独立的Docker容器中
- 集中式记忆：所有Agent共享一个记忆池（通过向量数据库）
- 可观测性：提供完整的执行日志、性能监控
核心代码（简化）
from langgraph.graph import StateGraph
from deerflow.runtime import DockerRuntime
# 定义多Agent工作流
graph = StateGraph(state_schema=dict)
# Agent 1：研究员工
graph.add_node("researcher", researcher_agent)
# Agent 2：写作者
graph.add_node("writer", writer_agent)
# Agent 3：审校者
graph.add_node("reviewer", reviewer_agent)
# 工作流：researcher → writer → reviewer
graph.add_edge("researcher", "writer")
graph.add_edge("writer", "reviewer")
# 在Docker沙箱中执行
runtime = DockerRuntime(image="python:3.12")
compiled_graph = graph.compile(runtime=runtime)
4.3 Claude Code：Anthropic的编程助手
Claude Code（Anthropic开发）是一个专注于编程任务的Agent。
架构特点
- 深度代码理解：Claude Code能够理解整个代码库（通过RAG + 代码AST分析）
- 多文件编辑：能够同时编辑多个文件，并保持一致性
- 测试驱动：自动编写测试用例，并验证代码正确性
- Git集成：自动commit、push到GitHub
核心技术
- 代码检索：基于向量相似度的代码搜索
- AST分析：用Tree-sitter解析代码结构
- 测试执行：集成pytest、jest等测试框架
第五章 性能优化：如何让Agent持续工作数周不"偏航"
长时程Agent面临的最大挑战是：如何在执行数周的任务中保持"不偏航"？
偏航的原因可能有：
- 目标漂移（Goal Drift）：执行过程中，Agent逐渐忘记了最初的目标
- 记忆污染（Memory Pollution）：积累了太多无关记忆，导致检索效率下降
- 错误累积（Error Accumulation）：早期的错误没有被纠正，导致后续所有步骤都错
- 陷入局部最优（Local Optimum）：Agent找到了一个"还行"的方案，就不再优化
5.1 防止目标漂移：目标对齐检查
解决方案：定期（比如每天）让Agent"回顾目标"，确保当前工作与最初目标一致。
def goal_alignment_check(agent, initial_goal: str):
"""目标对齐检查"""
prompt = f"""
初始目标：{initial_goal}
当前工作状态：
{agent.get_work_summary()}
请评估：
1. 当前工作是否仍然对齐初始目标？
2. 是否有偏离？如果有，偏离程度如