我用 300 行代码,实践了 Anthropic 的 Agent 架构哲学

张开发
2026/4/16 15:07:11 15 分钟阅读

分享文章

我用 300 行代码,实践了 Anthropic 的 Agent 架构哲学
代码不超过 300 行没有额外依赖实现了一套完整的 AI Agent 框架。 本文从设计哲学出发用真实代码讲清楚 Anthropic HarnessManaged Agents 到底在解决什么问题。缘起我为什么要重新发明轮子最近 AI Agent 框架满天飞——LangChain、AutoGen、CrewAI……每个都有几千个 star文档几百页。但我越用越困惑工具那么多为什么 Agent 还是那么脆直到我读了 Anthropic 内部的一篇设计文档里面有一句话让我停下来想了很久Harness 编码的是假设而假设会过时。这句话改变了我对 Agent 框架的理解。不是工具不够多而是我们一直在用错误的方式搭建 Agent——把太多假设硬编码进了架构里当模型能力提升后这些假设变成了枷锁。于是我花了一个周末用纯 Python 标准库无额外依赖实现了一套基于这个哲学的 Agent 框架叫做harness-agent。本文就是对这次实践的完整记录。先说结论三大哲学一套框架Anthropic 的 Managed Agents 架构建立在三个设计哲学之上哲学一为尚未被构想的程序设计接口保持稳定实现随时可换。不要把底层细节用哪个模型、工具怎么执行泄漏到上层。哲学二Harness 编码的是假设假设会过时你今天写的系统提示词、工具限制、迭代次数都是对模型能力的假设。模型在进步这些假设应该有机制被自动检测和放宽。哲学三大脑、双手、记忆彻底解耦控制逻辑大脑、工具执行双手、状态存储记忆——三者独立演化互不感知对方的实现细节。这三个哲学最终在代码里体现为两套东西三层架构解决怎么跑的技术问题Session / Harness / Sandbox三智能体 GAN 循环解决做什么的业务问题Planner / Generator / Evaluator让我们逐一拆解。第一层Session——记忆不属于大脑大多数 Agent 框架把对话历史存在 Agent 对象内部ounter(lineounter(lineounter(lineounter(lineounter(lineounter(lineounter(lineounter(lineounter(lineounter(line# 常见框架的做法class Agent: def __init__(self): self.history [] # 记忆硬绑定在 Agent 里 def chat(self, msg): self.history.append(msg) response llm.call(self.history) self.history.append(response) return response这有什么问题Agent 崩溃记忆就没了。Anthropic 的做法是把记忆彻底外置——Session 是独立于 Agent 存活的追加式事件日志ounter(lineounter(lineounter(lineounter(lineounter(lineounter(lineounter(lineounter(lineounter(lineounter(lineounter(lineounter(lineounter(lineounter(lineounter(lineounter(lineounter(lineounter(linedataclassclass Session: session_id: Optional[str] None # claude CLI 分配用于断点续传 events: List[Event] field(default_factorylist) cost_usd: float 0.0 trace_id: str field(default_factorylambda: uuid.uuid4().hex) # 跨 Agent 串联 def append(self, event: Event) - None: 追加事件。历史不可修改保证审计完整性。 self.events.append(event) def save(self, path: str) - None: 持久化为 JSONL每条事件一行 ... def load(cls, path: str) - Session: 从文件恢复支持断点续传 ...关键设计只增不改。事件一旦写入就不会被修改这保证了历史的可信度也让跨进程恢复成为可能ounter(lineounter(lineounter(lineounter(lineounter(lineounter(line# Agent 进程崩溃前保存session.save(/data/task_001.jsonl) # 新进程拉起后从断点继续session Session.load(/data/task_001.jsonl)harness.run(next_prompt, session) # 自动注入 --resume session_id这就是哲学三的第一个体现记忆层独立于大脑层。Session 不知道 Harness 用哪个模型Harness 不知道 Session 存在哪里——它们之间只有一个接口append(event)。第二层Harness——大脑是无状态的如果说 Session 是记忆那 Harness 就是大脑——但它是一个无状态的大脑。ounter(lineounter(lineounter(lineounter(lineounter(lineounter(lineounter(lineounter(lineounter(lineounter(lineounter(lineounter(lineounter(lineounter(lineounter(lineounter(lineounter(lineounter(lineounter(lineounter(lineounter(lineounter(lineounter(lineclass Harness: def run(self, prompt: str, session: Session) - Session: # 从 session 构建命令无状态所有状态来自 session cmd self._build_cmd(session) proc subprocess.Popen( cmd, stdinsubprocess.PIPE, # prompt 通过 stdin 传入 stdoutsubprocess.PIPE, stderrstderr_buf, ) proc.stdin.write(prompt) proc.stdin.close() # 解析事件流写入 session for line in proc.stdout: msg json.loads(line) if msg[type] assistant: session.append(Event(assistant, msg)) # 触发回调on_text / on_tool_call elif msg[type] result: session.session_id msg[session_id] session.cost_usd msg.get(total_cost_usd) or 0.0注意Harness 本身不存储任何状态所有状态都存在session里。这意味着同一个 Harness 实例可以驱动完全不同的任务也可以在任意时刻把 session 交给另一个进程接管。底层调用的是claude CLIounter(lineounter(lineounter(lineounter(lineclaude --print --output-format stream-json --verbose \ --permission-mode bypassPermissions \ --allowedTools Read,Bash,Glob \ --max-turns 50这里有一个关键发现bypassPermissions模式下--allowedTools白名单对工具调用的限制形同虚设——模型仍然可以调用任何工具只是不需要确认。这个发现让我意识到光靠 CLI 参数控制工具权限是不够的需要 Python 层的 Guardrail后面会讲。第三层Sandbox——双手不需要知道为什么Sandbox 是双手层它只做一件事告诉 Harness有哪些工具可以用怎么用。ounter(lineounter(lineounter(lineounter(lineounter(lineounter(lineounter(lineounter(lineclass Sandbox: def get_allowed_tools(self) - list: # 返回 [Bash, Read, Glob] # → 传给 Harness → --allowedTools 参数 def get_system_prompt_section(self) - str: # 返回工具能力的自然语言描述 # → 注入系统提示词让 Claude 知道如何使用这些工具Sandbox 不执行工具不感知业务逻辑只提供描述。工具的实际执行完全委托给claude CLI内置系统。这和 LangChain 那套工具是 Python 函数框架负责拦截调用的模式截然不同常见框架如 LangChainharness-agent Sandbox工具执行Python 函数框架拦截调用委托给 claude CLI工具注册注册表 类型系统自然语言描述工具限制拦截逻辑 权限检查allowed_tools 白名单维护成本高每个工具需要定义 Schema低哲学三的完整实现三层解耦接口清晰任意一层替换不影响其他层。护栏层当假设需要硬代码保障实践中我发现了一个尴尬的情况Planner 被设计为只读只允许 Read/Glob但在bypassPermissions模式下它仍然可以调用 Bash 写文件。光靠系统提示词说你只能读文件是不够的——模型偶尔会幻觉自己可以执行更多操作。这促使我加了一个Guardrail层在 Python 事件流里硬代码拦截ounter(lineounter(lineounter(lineounter(lineounter(lineounter(lineounter(linedataclassclass Guardrail: blocked_tools: list field(default_factorylist) # 工具黑名单 check_dangerous_bash: bool False # 拦截危险命令 check_path: bool False # 路径白名单 cwd: Optional[str] None action: GuardrailAction GuardrailAction.WARN # WARN / BLOCK10 条危险 Bash 命令模式rm -rf /、curl | bash、fork bomb 等6 条敏感信息正则API Key、私钥头部等在 Harness 事件流解析时同步触发ounter(lineounter(lineounter(lineounter(lineounter(line# Harness 内部if block[type] tool_use: if self.guardrail: result self.guardrail.check_tool_call(block[name], block[input]) self.guardrail.handle(result) # WARN 打印警告BLOCK 抛异常终止三个 Agent 预设不同的护栏ounter(lineounter(lineounter(lineplanner_guardrail(cwd) # 黑名单Agent, Bash, Edit, Write纯只读generator_guardrail(cwd) # 黑名单Agent检查危险 Bash 路径白名单evaluator_guardrail(cwd) # 黑名单Agent, Edit, Write验证只读不改产出这正是哲学一的体现Guardrail 的接口不变check_tool_call/check_text_output但规则可以随模型能力提升而放宽——不需要改 Harness 代码只需要调整 Guardrail 配置。三智能体 GAN把假设会过时变成机制现在来到最关键的部分。前面说的三层架构解决了怎么跑的问题但它不能回答我的 Agent 输出的东西到底好不好Anthropic 的答案是 GAN生成对抗三智能体架构ounter(lineounter(lineounter(lineounter(lineounter(lineounter(lineounter(lineounter(lineounter(lineounter(lineounter(line用户目标 ↓Planner规划一次 ↓ Sprint 合同含验收标准Generator执行每轮新上下文 ↓ 产出Evaluator验证独立上下文持怀疑态度 ↓ 评分 具体改进建议score ≥ pass_threshold ├─ 是 → 停止迭代 └─ 否 → 反馈给下一轮 Generator三个角色三种上下文策略Planner制定计划输出Sprint 合同——不是最终产物而是执行清单加验收标准。它只能读文件不能执行代码强制它做纯粹的规划者。Generator接收合同在全新上下文里执行。注意每次 Sprint 都清空上下文。这和大多数 Agent积累对话历史的做法相反——旧迭代的思维垃圾会污染新迭代的判断。Evaluator是最关键的角色。它在完全独立的上下文里运行不知道 Generator 说过什么只知道合同的验收标准和 Generator 的产出。它用工具实际验证而不是相信声明ounter(lineounter(lineounter(lineounter(lineounter(lineounter(lineounter(lineounter(lineounter(lineounter(lineounter(lineounter(lineounter(lineounter(lineounter(lineounter(line 验证quality_report.md 是否存在 → 文件存在大小 2.3KB 验证包含代码结构分析章节 → 找到代码结构分析标题 验证包含改进建议优先级排序 → 未找到优先级标记 ✅ 验证完成生成评分报告...{ overall_score: 7.2, passed: false, blocker: 改进建议缺少优先级排序, feedback: 第4节需要按严重程度对建议排序}分数低于pass_threshold默认 9.5反馈传给下一轮 Generator继续迭代。这就是哲学二的运行时实现Evaluator 在每次 Sprint 后自动检测Generator 的产出是否满足假设验收标准。当分数长期在第一次就达标时说明 pass_threshold 设得太低或者 Generator 能力已经超越了当初设计的约束——这是一个明确的信号是时候放宽假设了。可观测性trace_id 串联所有 Agent生产环境里最难调试的是一次 GAN 循环产生了几十条日志怎么把它们和一次用户请求关联起来解法是trace_id——在Orchestrator.run()入口处生成传递给所有子 Agent 的 Sessionounter(lineounter(lineounter(lineounter(lineounter(lineounter(line# orchestrator.pytrace_id session.trace_id # uuid4().hex32 位十六进制 plan_session Session(trace_idtrace_id) # Plannersprint_session Session(trace_idtrace_id) # Generatoreval_session Session(trace_idtrace_id) # Evaluator日志里过滤同一个trace_id就能还原完整的一次 GAN 任务流ounter(lineounter(lineounter(lineounter(lineounter(lineounter(lineounter(lineounter(line10:23:01 INFO orchestrator — run start trace_ida3f8c2d1 goal生成代码质量报告10:23:02 INFO planner — plan start trace_ida3f8c2d110:23:15 INFO generator — sprint start sprint1 trace_ida3f8c2d110:23:40 INFO evaluator — evaluate start sprint1 trace_ida3f8c2d110:23:55 INFO evaluator — evaluate done score7.2 passedFalse trace_ida3f8c2d110:23:55 INFO generator — sprint start sprint2 trace_ida3f8c2d110:24:20 INFO evaluator — evaluate done score9.6 passedTrue trace_ida3f8c2d110:24:20 INFO orchestrator — run done total_sprints2 cost$0.0312 trace_ida3f8c2d1这是 OpenTelemetry Span 的简化版实现——trace_id已经对齐了分布式追踪的概念未来接入 Jaeger/Zipkin 只需要在关键点埋点不需要改架构。评估历史让假设是否过时有数据支撑每次 Evaluator 打完分结果会写入 JSONL 文件ounter(lineounter(lineounter(lineounter(lineounter(lineounter(lineounter(lineounter(lineounter(lineounter(line# eval_history.pydataclassclass EvalRecord: timestamp: float task_id: str # goal 的前40字符 sprint: int score: float passed: bool blocker: str cost_usd: float任务结束后自动打印趋势和统计ounter(lineounter(lineounter(lineounter(lineounter(lineounter(line[Evaluator 趋势] ↑ 提升 2.4 分7.2 → 9.6 评估历史摘要 总评估次数: 8 通过次数: 5/862.5% 平均评分: 8.3 最高: 9.6 最低: 6.1 累计费用: $0.24当你看到通过率 62.5%平均评分 8.3并且知道 pass_threshold 是 9.5这时你就有数据决策是调低阈值放宽假设还是改进 Generator 的系统提示词还是扩大 Sandbox 的工具集。这就是哲学二从定性走向定量的路径。整体架构鸟瞰ounter(lineounter(lineounter(lineounter(lineounter(lineounter(lineounter(lineounter(lineounter(lineounter(lineounter(lineounter(lineounter(lineounter(lineounter(lineounter(lineounter(lineounter(lineounter(line┌──────────────────────────────────────────────────────────────┐│ Harness Engineering 四层 ││ ││ ┌──────────┐ ┌──────────┐ ┌────────────┐ ┌───────────┐ ││ │ 评估层 │ │ 护栏层 │ │ 可观测性层 │ │ 管理层 │ ││ │eval_hist │ │guardrail │ │ trace_id │ │orchestrat │ ││ │batch_eval│ │Guardrail │ │ logging │ │ EvalHist │ ││ └──────────┘ └──────────┘ └────────────┘ └───────────┘ ││ ││ 建立在三层架构之上 ││ Session记忆 Harness大脑 Sandbox双手 │└──────────────────────────────────────────────────────────────┘ │ ▼ 三智能体建立在三层之上┌──────────────────────────────────────────────────────────────┐│ GAN 三智能体业务逻辑层 ││ Planner规划→ Generator执行↔ Evaluator验证 ││ 每个 Agent 内部都是一个独立的三层架构实例 │└──────────────────────────────────────────────────────────────┘文件清单全部标准库零额外依赖ounter(lineounter(lineounter(lineounter(lineounter(lineounter(lineounter(lineounter(lineounter(lineounter(lineounter(linesession.py 记忆层追加式事件日志harness.py 大脑层无状态控制回路sandbox.py 双手层工具能力定义guardrail.py 护栏层硬代码工具拦截planner.py GAN规划器generator.py GAN生成器evaluator.py GAN评估器orchestrator.py GAN编排器eval_history.py 评估历史持久化batch_eval.py 离线批量评估CI 集成main.py 入口四种运行模式运行效果ounter(lineounter(lineounter(lineounter(lineounter(lineounter(lineounter(lineounter(lineounter(line# 最简单的三层架构演示python3 main.py --mode simple # 输出 [Sandbox/Simple] 工具白名单: Read, Bash, Glob [Session/Simple] 记录 assistant 事件共 1 条 [Harness/Simple] → Read: session.py [Harness/Simple] → Glob: **/*.py [Session/Simple] 记录 result 事件共 3 条 | 本次费用 $0.0018ounter(lineounter(lineounter(lineounter(lineounter(lineounter(lineounter(lineounter(lineounter(lineounter(lineounter(lineounter(lineounter(lineounter(lineounter(lineounter(lineounter(lineounter(lineounter(lineounter(lineounter(lineounter(lineounter(lineounter(lineounter(lineounter(lineounter(lineounter(lineounter(lineounter(lineounter(lineounter(lineounter(lineounter(lineounter(lineounter(lineounter(lineounter(lineounter(lineounter(line# GAN 三智能体完整运行python3 main.py --mode gan --pass-threshold 9.0 --max-iterations 3 --log-level INFO # 输出════════════════════════════════════════════════════════════ GAN 三智能体启动 目标: 分析当前目录的 Python 代码生成一份详细的代码质量...════════════════════════════════════════════════════════════ ╔══════════════════════════════╗║ Planner生成执行计划... ║╚══════════════════════════════╝ [Planner] → Read: harness.py [Planner] → Glob: **/*.py [Planner] → Read: session.py[Planner] 计划生成完成1847 字符 ╔══════════════════════════════════════╗║ GeneratorSprint #1 开始执行... ║╚══════════════════════════════════════╝ ⚙ 步骤 1读取所有 Python 文件 [Generator] → Glob: **/*.py ✓ 步骤 1 完成找到 11 个 Python 文件 ╔════════════════════════════════════════════╗║ Evaluator评估 Sprint #1 输出... ║╚════════════════════════════════════════════╝ 验证quality_report.md 是否存在 → 文件存在大小 3.1KB 验证代码质量评分章节 → 找到代码质量评分标题和评分表✅ 验证完成生成评分报告... [Orchestrator] ✓ 评分达标9.3 ≥ 9.0停止迭代 评估历史摘要 总评估次数: 1 通过次数: 1/1100% 平均评分: 9.3 最高: 9.3我学到了什么实现这套框架让我对Agent 为什么难做有了新的理解。问题不是工具不够而是层次混乱。当你把记忆、控制、执行塞进同一个 Agent 类里任何一个维度的变化都会牵动整体。模型升级了你需要改 Agent换工具了你需要改 Agent要加监控还是改 Agent。问题也不是缺少 Eval而是 Eval 没有闭环。大多数框架的 Eval 是离线的、事后的——你跑完一批测试看看对不对。但 Anthropic 的思路是把 Eval 变成 Agent 执行循环的一部分每次产出后立即评估评估结果反馈给下一轮执行。这是从开环到闭环的本质跨越。假设会过时是最深刻的一句话。你今天给 Agent 加的每一条约束都是你对当前模型能力的假设。把这些假设写死在代码里等模型升级后你就被过去的自己绑住了。Evaluator 的存在就是让系统有能力在运行时自动检测哪些假设还成立哪些可以放宽。如果你也在思考Agent 框架应该长什么样欢迎一起探讨。本文基于 Anthropic HarnessManaged Agents 设计哲学的实践总结。代码实现参考了 Anthropic 内部工程实践文档和 claude CLI 的 stream-json 协议。——新书推荐——Golang并发编程之美

更多文章