从50行代码到多智能体系统（非常详细），看懂Claude Code这篇就够了！

导语当你还在为写一个AI助手而苦恼时有人已经将Claude Code的架构拆解成12个渐进式模块。这不是简单的代码复现而是一场关于AI智能体本质的认知革命。今天我们将深入这个MIT开源项目的核心揭开生产级AI智能体系统的神秘面纱。一、颠覆认知模型就是智能体代码只是操控系统在开始之前让我们抛出一个可能颠覆你认知的观点AI智能体不是框架不是提示词链也不是拖拽式工作流——它是一个经过训练能够感知、推理和行动的深度神经网络。而你作为工程师的工作是构建让这个智能体有效运作的环境工具、知识、上下文管理、权限和观察接口。这个理念来自ShareAI Lab开源的Learn Claude Code项目它将Claude Code的架构逆向工程成12个渐进式的harness操控系统工程会话。每个会话都在一个不变的核心循环上添加恰好一个harness机制让你能够在看到它们组合之前孤立地理解每个概念。让我用一张图来说明这个核心架构可变的Harness层 不变的核⼼ tool_use end_turn ⽤⼾提⽰ messages数组 LLM调⽤ stop_reason? 执⾏⼯具 追加结果 返回响应 s01: 智能体循环 s02: ⼯具分发 s03: TodoWrite s04: ⼦智能体 s05: 技能加载 s06: 上下⽂压缩 s07-s12: ⾼级机制看到了吗中心的那个循环从s01到s12完全不变。变化的是围绕它的一切——赋予模型更丰富感知、更强大工具和更复杂协调能力的harness机制。二、架构全景12层Harness如何构建生产级系统让我们深入这12个会话看看它们如何从50行的单工具脚本演变成一个具有工作树隔离和自主任务认领的多智能体系统。2.1 基础层智能体循环与工具分发s01-s02s01智能体循环——一切的起点这个会话只有50行代码却包含了整个系统的DNA。核心逻辑极其简单while True: response llm.call(messages) if response.stop_reason tool_use: tool_result execute_tool(response.tool) messages.append({role: tool, content: tool_result}) elif response.stop_reason end_turn: return response.content就是这样一个while循环检查stop_reason执行工具追加结果继续循环。这个模式在整个项目中从未改变。s02工具分发——从单一到多元s02引入了关键的dispatch模式TOOLS { bash: bash_handler, read_file: read_file_handler, write_file: write_file_handler}def execute_tool(tool_call): handler TOOLS[tool_call.name] return handler(tool_call.arguments)这个name-to-handler的映射模式成为后续所有会话的支柱。每当你需要添加新工具时只需在TOOLS字典中注册一个新条目无需触碰核心循环逻辑。2.2 认知层规划与知识管理s03-s05s03TodoWrite——先计划后行动这是第一个真正体现智能的harness机制。TodoWrite工具强制模型在行动前先制定计划class TodoManager: def __init__(self): self.todos [] self.current_index 0 def create(self, items): self.todos [{id: i, text: item, status: pending} for i, item in enumerate(items)] def mark_complete(self, todo_id): self.todos[todo_id][status] completed self._check_and_nag() def _check_and_nag(self): pending [t for t in self.todos if t[status] pending] if not pending and self.todos: return ✅ All tasks completed! return f⚠️ {len(pending)} tasks remaining这个机制的价值在于它将隐式的思考过程显式化让模型必须先将复杂任务分解为可执行的子任务然后逐一完成。这不仅提高了任务完成率还提供了可观察的执行轨迹。s04子智能体——上下文隔离的艺术当任务过于复杂时单个智能体的上下文窗口会成为瓶颈。s04引入了子智能体模式class Subagent: def __init__(self, parent_messages): self.messages [] # 全新的messages数组 self.parent_messages parent_messages def execute(self, task_description): self.messages.append({ role: system, content: fYou are a subagent. Task: {task_description} }) # 运行独立的agent loop while True: response llm.call(self.messages) if response.stop_reason end_turn: return response.content # ... 处理工具调用关键点在于每个子智能体都有自己独立的messages数组。这意味着• 子任务不会污染父任务的上下文• 可以并行执行多个子任务• 每个子任务的上下文可以独立压缩和管理这是实现无限深度任务分解的基础。s05技能加载——按需注入领域知识与其在系统提示中塞入所有知识不如让模型按需加载def load_skill(skill_name): skill_path fskills/{skill_name}/SKILL.md with open(skill_path) as f: return f.read()def skill_loader_tool(skill_name): skill_content load_skill(skill_name) return ToolResult( contentskill_content, inject_to_next_promptTrue # 关键只注入到下一次调用 )这个机制的巧妙之处在于知识不占用初始上下文只有在需要时才加载临时注入通过tool_result返回只影响下一次LLM调用模块化每个技能是独立的markdown文件易于维护和更新项目提供了4个示例技能agent-builder、code-review、mcp-builder和pdf处理展示了这个模式的实际应用。2.3 扩展层上下文压缩与任务系统s06-s07s06上下文压缩——实现无限会话的关键随着对话进行messages数组会无限增长最终超出token限制。s06引入了三层压缩策略class ContextCompressor: def micro_compact(self, messages): 移除工具调用的详细参数只保留结果摘要 return [self._summarize_tool_call(m) if m[role] tool else m for m in messages] def auto_compact(self, messages, threshold0.8): 当token使用率达到阈值时自动压缩 if self.token_count(messages) threshold * MAX_TOKENS: return self._compact_older_messages(messages, keep_last10) return messages def full_summary(self, messages): 生成完整会话摘要替换历史记录 summary llm.generate_summary(messages) return [{role: system, content: fSession summary: {summary}}]三层策略Micro-compact移除工具调用的详细参数只保留结果Auto-compact当token使用率达到80%时压缩早期消息Full summary调用LLM生成完整会话摘要替换整个历史记录这个机制让智能体可以运行无限长的会话而不会超出上下文窗口限制。s07任务系统——文件级的持久化与依赖管理将任务从内存移到文件系统实现• 跨会话持久化• 任务依赖图• 多智能体协作class TaskSystem: def __init__(self, tasks_dir./tasks): self.tasks_dir tasks_dir os.makedirs(tasks_dir, exist_okTrue) def create(self, task_id, description, dependencies[]): task_file f{self.tasks_dir}/{task_id}.json task { id: task_id, description: description, status: pending, dependencies: dependencies, created_at: datetime.now().isoformat(), result: None } with open(task_file, w) as f: json.dump(task, f, indent2) def can_execute(self, task_id): 检查依赖是否全部完成 task self.load(task_id) for dep_id in task[dependencies]: dep_task self.load(dep_id) if dep_task[status] ! completed: return False return True def get_executable_tasks(self): 获取所有可执行的任务依赖已满足 all_tasks self.list_all() return [t for t in all_tasks if t[status] pending and self.can_execute(t[id])]这个任务系统形成了一个有向无环图DAG智能体可以根据依赖关系自动调度任务执行顺序。2.4 并发层后台任务与团队协作s08-s10s08后台任务——异步执行与通知引入daemon线程实现长时间运行的后台任务import threadingfrom queue import Queueclass BackgroundTaskManager: def __init__(self): self.tasks {} self.notification_queue Queue() def start_background_task(self, task_id, func, *args): def wrapper(): try: result func(*args) self.tasks[task_id] {status: completed, result: result} except Exception as e: self.tasks[task_id] {status: failed, error: str(e)} finally: self.notification_queue.put({ type: task_completed, task_id: task_id }) thread threading.Thread(targetwrapper, daemonTrue) thread.start() self.tasks[task_id] {status: running} return thread def check_notifications(self): 在主循环中检查后台任务完成通知 while not self.notification_queue.empty(): notification self.notification_queue.get() self._handle_notification(notification)这个机制允许智能体• 启动长时间运行的任务如模型训练、大规模数据处理• 继续处理其他任务而不阻塞• 在后台任务完成时收到通知并做出响应s09智能体团队——JSONL邮箱的异步通信多智能体协作的核心是通信机制。s09引入了基于JSONL文件的邮箱系统class AgentMailbox: def __init__(self, agent_id, team_dir./team): self.agent_id agent_id self.inbox_file f{team_dir}/{agent_id}_inbox.jsonl self.outbox_file f{team_dir}/{agent_id}_outbox.jsonl def send(self, to_agent_id, message_type, content): 发送消息到队友的邮箱 message { from: self.agent_id, to: to_agent_id, type: message_type, content: content, timestamp: datetime.now().isoformat(), read: False } outbox_file f./team/{to_agent_id}_inbox.jsonl with open(outbox_file, a) as f: f.write(json.dumps(message) \n) def receive(self): 读取未读消息 messages [] if os.path.exists(self.inbox_file): with open(self.inbox_file) as f: for line in f: message json.loads(line) if not message[read]: message[read] True messages.append(message) # 重写邮箱文件标记已读 self._rewrite_inbox(messages) return messagesclass AgentTeam: def __init__(self, agent_ids): self.agents { agent_id: Agent(agent_id, mailboxAgentMailbox(agent_id)) for agent_id in agent_ids } def run_team_cycle(self): 一轮团队协作 for agent in self.agents.values(): # 检查邮箱 messages agent.mailbox.receive() # 根据消息和自身状态决策 actions agent.process_messages(messages) # 执行动作可能包括发送消息给队友 for action in actions: if action[type] send_message: agent.mailbox.send( action[to], action[message_type], action[content] )JSONLJSON Lines格式的优势• 追加写入无需锁文件• 每行独立易于解析• 天然支持消息队列语义• 可持久化重启后不丢失s10团队协议——状态机驱动的协作仅仅有通信机制还不够智能体需要遵循协议。s10引入了两个关键的状态机1. 关闭握手协议class ShutdownProtocol: def __init__(self, team_size): self.shutdown_requests set() self.team_size team_size self.state running # running → shutting_down → stopped def request_shutdown(self, agent_id): self.shutdown_requests.add(agent_id) if len(self.shutdown_requests) self.team_size: self.state shutting_down return True # 全员同意可以关闭 return False def confirm_shutdown(self): self.state stopped2. 计划审批有限状态机FSMclass PlanApprovalFSM: def __init__(self): self.state idle # idle → proposed → approved → executing → completed self.current_plan None self.approvals set() def propose_plan(self, agent_id, plan): if self.state idle: self.current_plan plan self.proposer agent_id self.approvals {agent_id} self.state proposed return True return False def approve_plan(self, agent_id): if self.state proposed: self.approvals.add(agent_id) if len(self.approvals) self.required_approvals: self.state approved return True return False def start_execution(self): if self.state approved: self.state executing return True return False这些协议确保了团队协作的有序性和一致性避免了竞态条件和冲突。2.5 自主层自主代理与工作树隔离s11-s12s11自主代理——空闲循环与自动认领这是从被动响应到主动行动的关键跃迁class AutonomousAgent: def __init__(self, agent_id, task_system): self.agent_id agent_id self.task_system task_system self.state idle # idle → working self.current_task None def idle_cycle(self): 空闲时的自动行为 if self.state idle: # 检查任务板寻找可执行的任务 executable_tasks self.task_system.get_executable_tasks() if executable_tasks: # 自动认领第一个可用任务 task executable_tasks[0] self._claim_task(task) def _claim_task(self, task): 认领任务原子操作 if self.task_system.claim(task[id], self.agent_id): self.current_task task self.state working self._execute_task(task) def _execute_task(self, task): 执行任务 try: result self.run_agent_loop(task[description]) self.task_system.complete(task[id], result) finally: self.state idle self.current_task None关键机制空闲检测定期检查是否有可执行的任务原子认领使用文件锁确保任务不会被多个代理同时认领状态管理idle ↔ working状态转换自动释放任务完成后自动回到空闲状态这个模式实现了去中心化的任务调度无需中央协调器多个代理可以自主协作完成任务。s12工作树隔离——目录级的并行执行当多个代理同时操作文件系统时冲突不可避免。s12引入了Git worktree隔离import subprocessimport tempfileimport shutilclass WorktreeIsolation: def __init__(self, repo_path): self.repo_path repo_path self.worktrees {} def create_worktree(self, agent_id, branch_name): 为代理创建独立的工作树 worktree_path f./worktrees/{agent_id}_{branch_name} os.makedirs(worktree_path, exist_okTrue) # 创建新分支和工作树 subprocess.run([ git, worktree, add, worktree_path, -b, f{agent_id}_{branch_name} ], cwdself.repo_path, checkTrue) self.worktrees[agent_id] worktree_path return worktree_path def merge_worktree(self, agent_id, target_branchmain): 合并代理的工作树到主分支 worktree_path self.worktrees[agent_id] # 提交更改 subprocess.run([git, add, .], cwdworktree_path, checkTrue) subprocess.run([git, commit, -m, Auto-commit from agent], cwdworktree_path, checkTrue) # 合并到主分支 subprocess.run([ git, merge, f{agent_id}_branch, -m, fMerge work from {agent_id} ], cwdself.repo_path, checkTrue) # 清理工作树 subprocess.run([git, worktree, remove, worktree_path], cwdself.repo_path, checkTrue) del self.worktrees[agent_id]class IsolatedAgent(AutonomousAgent): def __init__(self, agent_id, task_system, worktree_manager): super().__init__(agent_id, task_system) self.worktree_manager worktree_manager self.worktree_path None def claim_task(self, task): # 认领任务时创建工作树 if super().claim_task(task): self.worktree_path self.worktree_manager.create_worktree( self.agent_id, task[id] ) return True return False def complete_task(self, result): # 完成任务时合并并清理工作树 try: if self.worktree_path: self.worktree_manager.merge_worktree(self.agent_id) finally: super().complete_task(result)工作树隔离的优势完全的文件系统隔离每个代理在独立的目录中工作版本控制Git自动跟踪所有更改冲突检测合并时Git会自动检测冲突可回滚任何问题都可以通过Git回滚这个机制让多个代理可以真正并行地修改同一个代码库而不会相互干扰。三、终极整合741行的生产级参考实现s_full.py将s01-s11的所有机制整合到一个741行的完整智能体中提供了完整的REPL命令集class FullAgent: def __init__(self): # s01: 核心循环 self.messages [] # s02: 工具分发 self.tools self._initialize_tools() # s03: 任务管理 self.todo_manager TodoManager() # s04: 子智能体 self.subagent_factory SubagentFactory() # s05: 技能加载 self.skill_loader SkillLoader() # s06: 上下文压缩 self.compressor ContextCompressor() # s07: 任务系统 self.task_system TaskSystem() # s08: 后台任务 self.background_manager BackgroundTaskManager() # s09-s10: 团队协作 self.team AgentTeam([agent_1, agent_2, agent_3]) # s11: 自主模式 self.autonomous_mode False def repl(self): 交互式命令行 while True: user_input input( ) if user_input.startswith(/): self._handle_command(user_input) else: self._handle_user_message(user_input) def _handle_command(self, command): if command /compact: self.messages self.compressor.auto_compact(self.messages) print(f✅ Context compacted. Token count: {self.token_count()}) elif command /tasks: tasks self.task_system.list_all() self._print_tasks(tasks) elif command /team: self._show_team_status() elif command /inbox: messages self.mailbox.receive() self._print_messages(messages) elif command /autonomous on: self.autonomous_mode True print( Autonomous mode enabled)这个参考实现展示了如何将所有harness机制有机整合形成一个生产就绪的智能体系统。四、架构启示为什么这个设计如此重要4.1 关注点分离的极致整个架构最精妙之处在于核心循环与harness机制的完全分离核心循环不变 while True: response llm(messages) if tool_use: execute_tool() if end_turn: returnHarness机制可变 - 工具集合TOOLS字典 - 上下文管理压缩策略 - 知识注入技能加载 - 协作协议邮箱、状态机这意味着•升级工具不影响核心逻辑•添加新功能只需扩展harness•测试和调试更加简单4.2 渐进式复杂度的工程智慧从50行到741行每一步只增加一个概念s01: 50行 - 核心循环s02: 80行 - 工具分发s03: 120行 - 任务规划s04: 160行 - 子智能体...s12: 900行 - 完整系统这种设计让学习者可以逐步理解每个session独立可运行孤立调试问题定位更精确按需学习根据需求选择学习深度4.3 文件系统作为状态存储整个系统大量使用文件系统而非数据库•任务持久化JSON文件•团队通信JSONL邮箱•上下文存储messages数组序列化•版本控制Git worktree优势• 无需数据库依赖• 天然可观察cat/less即可查看• 易于备份和迁移• 支持分布式NFS等共享存储五、实战应用你可以用这个架构做什么5.1 智能代码审查系统结合s05技能加载和s09团队协作# 加载代码审查技能skill skill_loader.load(code-review)# 创建审查团队team AgentTeam([reviewer_1, reviewer_2, security_expert])# 每个reviewer独立审查不同模块for module in codebase.modules: task_system.create( task_idfreview_{module}, descriptionfReview {module} for bugs and security issues, dependencies[] )# 自主代理自动认领审查任务agent_team.run_autonomous_review_cycle()5.2 自动化数据处理流水线利用s08后台任务和s12工作树隔离# 启动多个数据处理代理agents [ IsolatedAgent(fagent_{i}, task_system, worktree_manager) for i in range(10)]# 每个代理处理不同的数据分片for shard in data_shards: task_system.create( task_idfprocess_{shard.id}, descriptionfProcess data shard {shard.id}, dependencies[fvalidate_{shard.id}] )# 后台运行完成后通知background_manager.start_background_task( data_pipeline, agents.run_parallel_processing)5.3 智能客服机器人集群结合s07任务系统和s11自主代理# 创建客服代理集群support_agents [ AutonomousAgent(fsupport_{i}, task_system) for i in range(20)]# 客户问题作为任务进入系统def create_support_ticket(customer_id, issue): task_system.create( task_idfticket_{customer_id}, descriptionissue, prioritycalculate_priority(issue), dependencies[] )# 代理自动认领并处理工单while True: for agent in support_agents: agent.idle_cycle() # 自动认领可用工单 time.sleep(1)六、未来展望AI智能体工程的下一个十年这个架构揭示了一个重要趋势AI智能体开发正在从提示词工程转向系统工程设计。6.1 从单智能体到多智能体社会未来的AI系统不再是单个超级智能体而是由成百上千个 specialized agents专业代理组成的社会•分工协作每个代理专注于特定领域•自主协调通过协议和邮箱自主协作•容错性单个代理失败不影响整体6.2 Harness工程成为核心技能正如这个项目的核心理念模型就是智能体代码就是harness。未来工程师的核心竞争力将是• 设计高效的工具系统• 构建鲁棒的协作协议• 管理复杂的上下文和知识• 实现可扩展的调度机制6.3 开源架构推动行业标准化Learn Claude Code这样的开源项目将•降低入门门槛从0到1的学习路径清晰•建立最佳实践经过验证的架构模式•促进生态发展社区贡献工具和技能七、如何开始你的Harness工程之旅7.1 学习路径建议第一周理解核心git clone https://github.com/shareAI-lab/learn-claude-codecd learn-claude-codepython agents/s01_agent_loop.pypython agents/s02_tool_use.pypython agents/s03_todo_write.py第二周掌握上下文管理python agents/s04_subagent.pypython agents/s05_skill_loading.pypython agents/s06_context_compact.py第三周构建多智能体系统python agents/s07_task_system.pypython agents/s08_background_tasks.pypython agents/s09_agent_teams.py第四周实现生产级系统python agents/s10_team_protocols.pypython agents/s11_autonomous_agents.pypython agents/s_full.py # 终极挑战7.2 实践项目建议个人助手从s03开始构建一个能管理你待办事项的智能体代码审查员结合s05和s07创建一个自动代码审查系统数据分析团队使用s09-s11构建多代理数据分析流水线开源贡献为项目添加新的工具或技能模块7.3 进阶资源•官方文档项目提供中、英、日三语文档•Web平台Next.js交互式学习平台可视化每一步的执行流程•社区加入Discord/GitHub Discussions与其他harness工程师交流结语重新定义AI智能体开发当我们回顾这12个session的演进路径会发现一个惊人的事实智能体的智能并非来自复杂的框架而是来自对基础机制的精心编排。• 一个不变的while循环• 一个工具分发字典• 几个状态机• 一些JSONL文件• 加上Git worktree就构成了一个可以自主协作、并行执行、无限扩展的智能体系统。这就是Harness工程的精髓不是让代码更智能而是为智能创造最佳运作环境。正如这个项目的核心理念所言The model is the agent; the code is the harness.模型就是智能体代码就是操控系统。现在轮到你了。克隆仓库运行s01见证一个50行的循环如何演变成改变世界的智能体系统。学AI大模型的正确顺序千万不要搞错了2026年AI风口已来各行各业的AI渗透肉眼可见超多公司要么转型做AI相关产品要么高薪挖AI技术人才机遇直接摆在眼前有往AI方向发展或者本身有后端编程基础的朋友直接冲AI大模型应用开发转岗超合适就算暂时不打算转岗了解大模型、RAG、Prompt、Agent这些热门概念能上手做简单项目也绝对是求职加分王给大家整理了超全最新的AI大模型应用开发学习清单和资料手把手帮你快速入门学习路线:✅大模型基础认知—大模型核心原理、发展历程、主流模型GPT、文心一言等特点解析✅核心技术模块—RAG检索增强生成、Prompt工程实战、Agent智能体开发逻辑✅开发基础能力—Python进阶、API接口调用、大模型开发框架LangChain等实操✅应用场景开发—智能问答系统、企业知识库、AIGC内容生成工具、行业定制化大模型应用✅项目落地流程—需求拆解、技术选型、模型调优、测试上线、运维迭代✅面试求职冲刺—岗位JD解析、简历AI项目包装、高频面试题汇总、模拟面经以上6大模块看似清晰好上手实则每个部分都有扎实的核心内容需要吃透我把大模型的学习全流程已经整理好了抓住AI时代风口轻松解锁职业新可能希望大家都能把握机遇实现薪资/职业跃迁这份完整版的大模型 AI 学习资料已经上传CSDN朋友们如果需要可以微信扫描下方CSDN官方认证二维码免费领取【保证100%免费】