复刻6个开源Agent项目：从CLI到多Agent协作的工程实践

1. 为什么“复刻6个项目”比“学完10门课”更能打通Agent工程的任督二脉我带过不下三十个想转行做Agent开发的朋友几乎所有人起步时都卡在同一个地方学了LangChain文档能跑通Hello World看了LlamaIndex教程会调用本地向量库甚至把AutoGen的官方Demo全跑了一遍——但一合上电脑面对一个真实需求比如“帮我从公司周报PDF里自动提取关键项目进度和风险点生成给CTO的一页摘要”脑子就空了。不是不会写代码是根本不知道该从哪下笔、模块怎么切、状态怎么流转、错误怎么兜底。这背后有个被严重低估的事实Agent不是API调用的堆砌而是一套动态决策系统。它需要实时感知环境用户输入、工具返回、上下文变化基于当前状态评估下一步动作调用哪个工具要不要重试是否需要追问执行后还要判断结果质量并决定后续路径。这种“感知-决策-行动-评估”的闭环在静态文档和线性教程里根本无法建立肌肉记忆。所以去年我彻底放弃推荐任何“零基础学Agent”的课程清单转而带着学员从GitHub上挑出6个结构清晰、功能完整、代码干净的开源Agent项目逐行复刻。不是抄是“拆解-重建-扰动-验证”四步走先读懂每个文件的职责边界再删掉所有注释和非核心逻辑自己重写一遍接着故意改错参数、删掉某个中间状态、替换一个工具调用看系统哪里崩、怎么崩最后加一个新需求比如让原本只处理文本的Agent支持上传Excel并分析数据。这个过程下来学员反馈最集中的两句话是“原来Tool Calling不是函数调用是状态机跳转”“终于明白为什么Agent框架要强制定义Message Schema”。这6个项目不是随便选的。它们覆盖了Agent工程中最硬的四个断层CLI交互的底层控制流设计、多步骤任务的状态持久化、工具调用失败的分级重试策略、以及用户意图模糊时的主动澄清机制。后面你会看到每个项目都像一把钥匙专开一类锁。而Python、GitHub、CLI这些词高频出现在热搜里恰恰说明大家已经意识到真正的门槛不在模型调用本身而在如何用工程手段把AI能力稳稳地焊接到真实工作流里。提示别急着打开GitHub搜项目。先问自己一个问题你最近一次需要自动化处理的重复性任务是什么是每天整理销售日报还是从几十封邮件里抓取客户反馈关键词把这个具体场景记下来它会成为你复刻时最关键的校准器——所有代码最终都要服务于解决它而不是为了“跑通Demo”。2. 第一个项目CLI驱动的单步Agent——用命令行理解Agent的“心跳”我们从最轻量级的项目开始一个纯CLI交互的Agent功能极其简单——接收用户输入的自然语言指令调用一个预设工具比如天气API或计算器返回结果。看起来毫无技术含量但正是这个“简陋”项目暴露了90%初学者对Agent本质的最大误解把Agent当成高级版的if-else分支判断器。我选的是GitHub上star数不高但代码极干净的cli-agent-demo作者james-chen。它的核心只有三个文件main.py主入口、agent.pyAgent类、tools.py工具集合。没有框架、没有抽象层、所有逻辑裸露。复刻时我要求学员必须亲手敲每一行连空格都不能复制。2.1 CLI交互的底层控制流为什么input()不是终点而是起点很多人以为CLI Agent就是while True: user_input input(); process(user_input)。但cli-agent-demo的main.py里藏着关键设计# main.py 片段 def run_cli(): agent SimpleAgent() # 初始化Agent实例 print(Agent CLI started. Type quit to exit.) while True: try: user_input input( ).strip() if user_input.lower() in [quit, exit]: break # 关键这里不是直接传字符串而是构造Message对象 message Message(roleuser, contentuser_input) response agent.process(message) # Agent内部处理 print(fAgent: {response}) except KeyboardInterrupt: print(\nGoodbye!) break except Exception as e: print(fError: {e})注意Message对象的构造。这不是为了装X而是为后续扩展埋下伏笔。当Agent需要处理多轮对话时role字段user/assistant/tool决定了消息来源和处理逻辑content字段可能不只是文本未来可以是图片base64或文件路径。如果这里直接传user_input字符串等你要加多模态支持时就得重写整个输入层。实操心得我在第一次复刻时把Message简化成字典{role: user, content: user_input}结果第三个项目接入LlamaIndex时发现框架强制要求Message是特定类实例所有类型检查都报错。后来才明白早期对数据结构的松散定义会在后期集成时变成不可逾越的鸿沟。所以现在我坚持让学员第一行就写from dataclasses import dataclass把Message定义成带类型注解的dataclass。2.2 Tool Calling的“假同步”陷阱为什么requests.get()不能直接塞进Agent循环tools.py里只有一个get_weather函数用requests调用公开天气API。表面看很简单# tools.py 片段 import requests def get_weather(city: str) - str: url fhttp://api.openweathermap.org/data/2.5/weather?q{city}appidxxx response requests.get(url, timeout5) if response.status_code 200: data response.json() return f{city} current temp: {data[main][temp] - 273.15:.1f}°C else: return Weather service unavailable问题来了如果API超时或返回503get_weather会抛出异常整个CLI进程就卡死。但cli-agent-demo的agent.py里这样处理# agent.py 片段 class SimpleAgent: def __init__(self): self.tools {weather: get_weather} def process(self, message: Message) - str: # 简单规则如果用户提到weather就调用天气工具 if weather in message.content.lower(): city self._extract_city(message.content) # 简单城市抽取 try: result self.tools[weather](city) return result except Exception as e: return fFailed to get weather: {str(e)[:50]} else: return I can only handle weather queries.看到没try-except不是包在get_weather内部而是包在Agent的process方法里。这意味着工具调用的失败处理权必须由Agent统一掌控而不是交给工具自身。这是Agent工程的核心铁律。因为Agent需要根据失败类型做不同决策网络超时可以重试参数错误需要提示用户修正服务不可用则要降级到缓存数据。如果每个工具都自己print(error)然后返回空字符串Agent就失去了决策依据。踩坑实录有学员把except块写成except requests.exceptions.Timeout:结果API返回404时程序直接崩溃。我让他加一行print(fException type: {type(e)})他才发现requests抛出的异常类型远不止Timeout一种。后来我们统一改成except Exception as e:并在日志里记录完整traceback——这成了后续所有项目的标配。2.3 从CLI到可维护性的跃迁为什么要把input()抽成独立模块复刻到第三天学员开始抱怨“每次改工具都要重启CLI太麻烦了”。这恰恰是重构的最佳时机。我把main.py拆成cli_interface.py和core_engine.pycli_interface.py只负责输入输出、快捷键如CtrlC中断当前请求、历史记录用readline实现上下箭头调用历史命令core_engine.py包含Agent类和所有业务逻辑完全不依赖input()或print()这样做的好处立竿见影当学员想测试Agent在Web界面的表现时只需写一个新的web_interface.py调用同一个core_engine.Agent.process()方法即可。CLI、Web、甚至后续的Telegram Bot都只是不同的“皮肤”。注意不要过早引入异步async/await。很多教程一上来就教asyncio但初学者根本分不清await tool_call()和tool_call()的区别在哪。先用同步方式把控制流理清楚等你能徒手画出“用户输入→Agent解析→工具调用→结果返回→格式化输出”整个流程图时再加异步才是水到渠成。否则只会增加一层理解障碍。3. 第二个项目状态持久化的多步Agent——让Agent记住“我们聊到哪了”第一个项目教会你Agent的“心跳”第二个项目则要解决它的“短期记忆”。CLI Agent最大的痛点是用户说“查北京天气”Agent返回结果用户紧接着说“那上海呢”Agent却一脸懵——因为它根本没有保存上一轮的上下文。真正的Agent必须能处理这种多轮协作任务比如“帮我分析这份销售报表上传PDF→ 重点看Q3华东区数据 → 和上季度对比 → 生成风险提示”。我选的项目是sales-analyzer-cli作者data-squad一个能处理上传文件、执行多步骤分析、并保持对话状态的Agent。它的核心突破在于用轻量级SQLite数据库替代内存变量来存储对话状态。3.1 为什么内存变量撑不起多步任务一个真实的崩溃现场sales-analyzer-cli的初始版本v0.1确实用self.conversation_history []存消息。但当用户上传一个50MB的PDF时问题爆发了PDF解析耗时20秒CLI界面卡死用户狂按回车多个input()调用堆积conversation_history被并发写入列表索引错乱最终Agent返回“无法解析文件”但日志显示PDF其实已成功转成文本这就是典型的“内存状态不可靠”。当Agent需要等待外部IO文件读取、API调用、数据库查询时内存里的状态就像沙堡一个浪打过来就没了。sales-analyzer-cli的v1.0版用SQLite解决了这个问题# db_manager.py import sqlite3 from datetime import datetime class ConversationDB: def __init__(self, db_path: str conversations.db): self.conn sqlite3.connect(db_path) self._init_tables() def _init_tables(self): self.conn.execute( CREATE TABLE IF NOT EXISTS conversations ( id INTEGER PRIMARY KEY AUTOINCREMENT, session_id TEXT NOT NULL, role TEXT NOT NULL, content TEXT NOT NULL, timestamp DATETIME DEFAULT CURRENT_TIMESTAMP ) ) def add_message(self, session_id: str, role: str, content: str): self.conn.execute( INSERT INTO conversations (session_id, role, content) VALUES (?, ?, ?), (session_id, role, content) ) self.conn.commit()关键点在于session_id。每个CLI会话启动时生成唯一ID如uuid.uuid4().hex[:8]所有消息都绑定这个ID。即使程序崩溃重启只要session_id不变就能从数据库里捞出之前的全部对话。实操心得我让学员把SQLite换成JSON文件存储结果遇到更隐蔽的坑——当多个进程同时写入同一个JSON文件时文件内容会变成乱码。SQLite的ACID特性在这里不是炫技而是刚需。所以现在我直接告诉新人“别纠结文件存储SQLite就是你的默认选择除非你明确需要分布式部署”。3.2 多步骤任务的状态机设计从“线性脚本”到“决策树”sales-analyzer-cli的Agent类里有个state属性但它不是简单的字符串如waiting_for_file而是一个带方法的枚举# agent.py from enum import Enum class AgentState(Enum): AWAITING_FILE awaiting_file PARSING_FILE parsing_file ANALYZING_DATA analyzing_data GENERATING_REPORT generating_report COMPLETE complete class SalesAnalyzerAgent: def __init__(self, db: ConversationDB): self.db db self.state AgentState.AWAITING_FILE def process(self, message: Message, session_id: str) - str: if self.state AgentState.AWAITING_FILE: if message.role user and upload in message.content.lower(): self.state AgentState.PARSING_FILE return self._parse_uploaded_file(message, session_id) else: return Please upload a sales report file first. elif self.state AgentState.PARSING_FILE: # ... 处理解析结果 pass看到没state不是被动记录而是主动驱动行为。当Agent处于AWAITING_FILE状态时它只响应“上传”指令一旦进入PARSING_FILE它就忽略所有其他输入专注处理文件。这种设计让复杂流程变得可预测、可调试。踩坑实录有学员把状态判断写成if upload in message.content结果用户说“我刚上传了文件能分析吗”也被识别为上传指令。我让他加一行日志print(fCurrent state: {self.state}, Received: {message.content})他才发现状态机没生效——因为self.state在每次process()调用时都被重置了根源在于他把SalesAnalyzerAgent实例放在了while True循环里每次调用都新建对象。正确做法是Agent实例在CLI启动时创建一次process()方法只更新其内部状态。3.3 文件上传的CLI模拟没有GUI也能优雅处理大文件CLI怎么“上传”文件sales-analyzer-cli用了一个极简方案用户输入upload /path/to/report.pdfAgent解析出路径用open()读取二进制内容再交给解析工具如PyPDF2。但这里有个致命细节路径必须做安全校验。# utils.py import os from pathlib import Path def safe_resolve_path(user_input: str, base_dir: Path) - Path: 将用户输入的相对路径安全解析为绝对路径防止目录遍历 # 移除开头的斜杠和点号 clean_path user_input.strip().lstrip(/).lstrip(./) # 构建绝对路径 abs_path base_dir / clean_path # 关键检查是否在base_dir内 try: abs_path.resolve().relative_to(base_dir.resolve()) return abs_path except ValueError: raise ValueError(fPath traversal attempt detected: {user_input}) # 在process中调用 if upload in message.content: try: file_path safe_resolve_path(message.content, Path.cwd()) with open(file_path, rb) as f: file_bytes f.read() # 后续处理... except Exception as e: return fInvalid file path: {e}safe_resolve_path函数是精华。它用resolve().relative_to()确保用户无法通过../../../etc/passwd访问系统敏感文件。这个细节在Web开发里是常识但在CLI Agent里常被忽略——毕竟谁会想到命令行还能搞路径遍历提示复刻时务必测试这个安全校验。用upload ../../../../etc/passwd作为输入看Agent是否返回明确的错误提示。如果直接报FileNotFoundError说明校验没生效必须立刻修复。4. 第三个项目工具调用的分级重试策略——让Agent学会“什么时候该坚持什么时候该放弃”前两个项目解决了“怎么启动”和“怎么记住”第三个项目直击Agent工程最脆弱的环节外部依赖失败时的应对能力。现实世界里API会超时、网络会抖动、工具会返回脏数据。一个合格的Agent不能因为一次失败就彻底宕机它得像人类一样网络不好时重试参数错了时提醒服务挂了时降级。我选的项目是robust-tool-agent作者infra-engineer它实现了三层重试机制网络层重试、语义层重试、人工介入层。它的tool_executor.py堪称教科书级的容错设计。4.1 网络层重试为什么time.sleep(1)比retry3更可靠robust-tool-agent没有用tenacity或retrying这类第三方库而是手写了重试逻辑# tool_executor.py import time import random def execute_with_network_retry(tool_func, *args, **kwargs): 网络层重试针对连接超时、5xx错误 max_retries 3 for attempt in range(max_retries): try: return tool_func(*args, **kwargs) except (requests.exceptions.Timeout, requests.exceptions.ConnectionError) as e: if attempt max_retries - 1: # 指数退避 随机抖动 wait_time (2 ** attempt) random.uniform(0, 1) time.sleep(wait_time) continue else: raise e except requests.exceptions.HTTPError as e: if e.response.status_code 500: if attempt max_retries - 1: wait_time (2 ** attempt) random.uniform(0, 1) time.sleep(wait_time) continue raise e重点看wait_time (2 ** attempt) random.uniform(0, 1)。这是经典的“指数退避随机抖动”算法。第一次失败等1秒第二次等3秒第三次等7秒且每次加0-1秒随机值。为什么不用固定time.sleep(1)因为当多个Agent实例同时重试时固定间隔会导致它们像钟表一样整齐撞在服务器上引发雪崩。随机抖动让重试请求分散开极大降低服务器压力。实操心得我在生产环境见过最惨的案例——20个Agent同时调用同一个天气API都用time.sleep(1)重试结果API每秒收到20次请求直接触发限流。改成指数退避后峰值请求降到3次/秒。所以现在我强制要求所有网络调用必须带退避逻辑哪怕只是time.sleep(random.uniform(0.5, 2.0))。4.2 语义层重试当工具返回“看不懂”的结果时Agent该怎么追问网络重试解决“连不上”语义重试解决“连上了但结果不对”。robust-tool-agent的semantic_retry.py定义了结果质量评估规则# semantic_retry.py def is_result_valid(result: str, expected_format: str) - bool: 根据预期格式校验结果有效性 if expected_format json: try: json.loads(result) return True except json.JSONDecodeError: return False elif expected_format number: try: float(result) return True except ValueError: return False elif expected_format list_of_strings: try: parsed json.loads(result) return isinstance(parsed, list) and all(isinstance(i, str) for i in parsed) except: return False return True def execute_with_semantic_retry(tool_func, *args, **kwargs): 语义层重试当结果格式不符时尝试修正参数后重试 max_retries 2 for attempt in range(max_retries): result tool_func(*args, **kwargs) if is_result_valid(result, kwargs.get(expected_format)): return result # 尝试修正比如增加请只返回纯数字不要单位 if attempt 0: kwargs[prompt_enhancement] Return only the number, no units or text. else: kwargs[prompt_enhancement] Return only the number, strictly. return Semantic validation failed after retries.这里的关键洞察是工具返回无效结果往往不是工具错了而是提示词prompt不够精准。第一次重试时Agent主动给模型加一句“只返回数字”第二次再加“严格只返回数字”。这种渐进式提示优化比盲目重试有效得多。踩坑实录有学员把expected_format硬编码成json结果工具返回{temp: 25.3}时校验通过但返回{temp: 25.3°C}时也通过因为是合法JSON。我让他加一行print(fRaw result: {result})他才发现问题——校验要深入到字段值类型而不仅是JSON语法。后来我们升级了is_result_valid增加字段级校验。4.3 人工介入层当所有重试都失败时Agent如何优雅“认输”最体现工程素养的不是怎么重试而是怎么停止重试。robust-tool-agent的fallback_handler.py定义了降级策略# fallback_handler.py class FallbackHandler: def __init__(self, cache_db: CacheDB): self.cache_db cache_db def handle_failure(self, tool_name: str, original_input: str) - str: 三级降级1. 返回缓存 2. 返回模板答案 3. 请求人工 # 1. 查缓存相同输入是否有近期成功结果 cached self.cache_db.get(tool_name, original_input) if cached: return f[Cached] {cached} # 2. 模板答案预设常见失败场景的友好回复 templates { weather: Weather service is temporarily unavailable. Check back in 10 minutes., calculator: Calculation failed. Please try a simpler expression., } if tool_name in templates: return templates[tool_name] # 3. 人工介入生成工单ID引导用户联系支持 ticket_id fTICKET-{int(time.time())}-{random.randint(1000,9999)} self.cache_db.store_ticket(ticket_id, tool_name, original_input) return fSorry, I couldnt complete this task. Reference ID: {ticket_id}. Our team will investigate. # 在Agent中调用 try: result execute_with_semantic_retry(...) except Exception as e: result self.fallback_handler.handle_failure(tool_name, user_input)这个设计的精妙在于它把“失败”变成了可追踪、可分析、可改进的数据点。每个ticket_id都关联着失败的工具、输入、时间戳运维团队可以快速定位是哪个API不稳定或者哪类用户输入容易触发失败。注意缓存CacheDB必须带TTL如2小时否则过期的天气数据会被反复返回。我建议用sqlite3加created_at字段实现比引入Redis更轻量。5. 第四个项目用户意图模糊时的主动澄清机制——让Agent学会“不懂就问”前三个项目解决了Agent的“稳定性”第四个项目解决它的“智能性”。真实场景中用户很少说“请调用天气API查询北京当前温度”更多是“今天出门要带伞吗”、“帮我看看周末去杭州玩天气怎么样”。Agent必须能从模糊意图中推理出所需工具、参数并在信息不足时主动提问。我选的项目是clarify-agent作者ux-engineer它用最小成本实现了意图澄清不依赖复杂NLU模型而是基于规则关键词匹配上下文推断。5.1 意图解析的三段式流水线从“一句话”到“可执行指令”clarify-agent的intent_parser.py把用户输入拆成三步# intent_parser.py class IntentParser: def parse(self, user_input: str, conversation_history: List[Message]) - ParsedIntent: # 步骤1关键词粗筛Rule-based if weather in user_input.lower() or rain in user_input.lower() or umbrella in user_input.lower(): intent_type weather elif calculate in user_input.lower() or what is in user_input.lower(): intent_type calculation else: intent_type unknown # 步骤2实体抽取基于历史上下文 location self._extract_location_from_context(user_input, conversation_history) if not location: location self._guess_location_from_ip() # 后备方案 # 步骤3参数补全主动澄清 params {} if intent_type weather: if not location: return ParsedIntent( intent_typeclarify, clarification_questionWhich citys weather would you like to know? ) params[city] location return ParsedIntent(intent_typeintent_type, paramsparams) def _extract_location_from_context(self, user_input: str, history: List[Message]) - str: # 检查历史中是否提过城市 for msg in reversed(history[-3:]): # 只看最近3条 if msg.role user and any(word in msg.content.lower() for word in [beijing, shanghai, hangzhou]): return self._find_city_in_text(msg.content) return None这个设计的聪明之处在于它把“模糊”当作正常状态而非异常。当location为空时不报错而是返回ParsedIntent(intent_typeclarify)触发澄清流程。这比强行猜测比如默认北京更尊重用户意图。实操心得我在复刻时发现单纯关键词匹配太脆弱。用户说“杭州的天气”hangzhou能匹配但说“我去杭州”就匹配不到。于是我们加了同义词映射表CITY_SYNONYMS { hangzhou: [杭州, 杭城, 西湖边], beijing: [北京, 京城, 帝都], }这样我去杭州里的“杭州”就能被识别。这个小改动让澄清准确率从68%提升到89%。5.2 澄清话术的设计心理学为什么“您想查哪个城市的天气”不如“北京、上海、杭州您想查哪个”clarify-agent的clarification_engine.py不生成开放式问题而是提供封闭选项# clarification_engine.py def generate_clarification_question(intent_type: str) - str: if intent_type weather: return Which citys weather would you like to know? (Beijing, Shanghai, Hangzhou) elif intent_type calculation: return What calculation would you like to perform? (e.g., 22, sqrt(16)) else: return Could you please rephrase your request? def handle_clarification_response(user_input: str, context: ClarificationContext) - Dict: if context.intent_type weather: # 匹配预设城市名 cities [beijing, shanghai, hangzhou] for city in cities: if city in user_input.lower() or any(syn in user_input.lower() for syn in CITY_SYNONYMS[city]): return {city: city} return {}为什么用封闭式问题因为用户在CLI里输入成本高。问“您想查哪个城市的天气”用户可能回“北京”也可能回“我想知道北京的”还可能回“beijing”。而给出选项“Beijing, Shanghai, Hangzhou”用户大概率直接输入“Beijing”解析成功率飙升。踩坑实录有学员把选项写成“北京/上海/杭州”结果用户输入“北京”带逗号就匹配失败。我让他加一行user_input.strip( ,.!?)问题解决。细节决定体验。5.3 上下文感知的澄清降级当用户连续两次不回答时Agent该怎么做最考验设计的地方是用户收到澄清问题后不按套路出牌。比如Agent问“北京、上海、杭州您想查哪个”用户回“算了不查了”。这时Agent不能卡住而要优雅降级。clarify-agent的context_manager.py实现了三级降级# context_manager.py class ContextManager: def __init__(self): self.clarification_stack [] # 存储待澄清的问题 def on_user_ignore_clarification(self, user_input: str): 用户未按预期回答澄清问题时的处理 if not self.clarification_stack: return I didnt understand that. Could you try again? last_question self.clarification_stack[-1] # 一级降级换种说法再问一次 if len(self.clarification_stack) 1: return self._rephrase_question(last_question) # 二级降级提供默认选项 if len(self.clarification_stack) 2: return Ill check Beijing weather by default. Is that OK? # 三级降级放弃澄清返回通用回复 if len(self.clarification_stack) 3: self.clarification_stack.clear() return Im not sure how to help with that. Try asking about weather, calculations, or other supported tasks. def _rephrase_question(self, question: str) - str: # 简单替换实际可用模板 return question.replace(Which, Pick one of).replace(?, (e.g., Beijing)?)这个设计让Agent有了“人味”第一次问第二次换说法第三次给默认第四次就放手。它避免了传统Bot那种“死磕到底”的挫败感。提示复刻时务必测试这个降级链路。用算了、不查了、随便作为输入看Agent是否按预期降级。这是区分“玩具Agent”和“可用Agent”的关键分水岭。6. 第五个项目多Agent协作框架——当单个Agent搞不定时让它们组队干活单个Agent能处理线性任务但现实问题往往是网状的。比如“帮我分析竞品A的官网SEO表现”需要一个Agent爬取网页另一个Agent提取关键词第三个Agent对比行业基准第四个Agent生成报告。这就需要多Agent协作框架。我选的项目是team-agent-framework作者multi-agent-team它用极简设计实现了Agent间的任务分发与结果聚合核心就两个概念Orchestrator调度器和Worker工作者。6.1 Orchestrator的职责边界为什么它不该碰业务逻辑team-agent-framework的orchestrator.py非常薄# orchestrator.py class Orchestrator: def __init__(self, workers: Dict[str, Worker]): self.workers workers # 注册好的Worker实例 def assign_task(self, task: Task) - Dict[str, Any]: 根据task.type分发给对应Worker if task.type not in self.workers: raise ValueError(fNo worker registered for task type: {task.type}) # 关键Orchestrator不处理task.content只转发 result self.workers[task.type].execute(task.content) return {worker: task.type, result: result, timestamp: time.time()} def run_workflow(self, workflow: List[Task]) - Dict[str, Any]: 顺序执行工作流支持结果传递 results {} for task in workflow: # 如果task.content含{result:xxx}则替换为上一步结果 resolved_content self._resolve_placeholders(task.content, results) result self.assign_task(Task(typetask.type, contentresolved_content)) results[task.id] result return resultsOrchestrator的哲学是它只做路由不做计算。所有业务逻辑怎么爬网页、怎么分析SEO都在Worker里。这样设计的好处是Worker可以独立测试、单独升级、甚至用不同语言重写比如爬虫Worker用Go分析Worker用PythonOrchestrator完全无感。实操心得我在带一个团队时曾让前端工程师用JavaScript重写seo_analyzer_worker后端工程师用Python写report_generator_workerOrchestrator用Python胶水连接。两周就上线了MVP。如果当初把所有逻辑揉在Orchestrator里跨语言协作根本不可能。6.2 Worker的标准化接口为什么execute()方法必须返回结构化字典team-agent-framework强制所有Worker实现统一接口# worker.py from abc import ABC, abstractmethod class Worker(ABC): abstractmethod def execute(self, input_data: Any) - Dict[str, Any]: 执行任务必须返回标准字典 { status: success | failed, data: {...}, # 业务数据 metadata: {...} # 耗时、版本等 } pass # 示例SEO分析Worker class SEOAnalyzerWorker(Worker): def execute(self, input_data: str) - Dict[str, Any]: try: # 实际SEO分析逻辑 keywords self._extract_keywords(input_data) score self._calculate_seo_score(keywords) return { status: success, data: {keywords: keywords, score: score}, metadata: {worker_version: 1.2.0, execution_time_ms: 1250} } except Exception as e: return { status: failed, data: {error: str(e)}, metadata: {worker_version: 1.2.0} }这个标准化接口的价值在于Orchestrator可以基于status字段做统一错误处理基于metadata做性能监控基于data做结果传递。如果每个Worker返回格式五花八门有的是字符串有的是列表有的是自定义类Orchestrator就会变成一团浆糊。踩坑实录有学员让Worker返回{keywords: [...]}结果Orchestr