1. 为什么“调用一次LLM”不等于“跑通一个Agent”很多人第一次接触AI Agent时心里想的其实是“不就是把prompt发给大模型再把结果拿回来吗我昨天刚用curl调通了OpenAI API这不就算完成了”——这种想法非常普遍也特别危险。它就像刚学会拧螺丝就觉得自己能造汽车忽略了整个系统里真正决定成败的那些“看不见的零件”。我见过太多团队在Demo阶段用几行Python脚本硬编码prompt三分钟就跑出个“能聊天的Agent”结果一进真实业务场景就崩用户连续追问三次Agent开始胡编接口文档任务链里某个步骤失败整个流程卡死不报错多轮对话中上下文突然丢失前一秒还在查订单后一秒说“我不记得你之前问过什么”。这些不是LLM的问题而是架构选择失当的必然结果。核心矛盾在于单次LLM调用是原子操作而Agent是状态驱动、可中断、可回溯、需协同的运行时系统。它必须处理的远不止“输入→模型→输出”这一条直线。比如当用户说“帮我对比A和B两款手机并生成购买建议”这个请求背后至少要拆解为检索A/B参数→提取关键指标→调用计算器比对性能分→查电商平台实时价格→综合生成建议。这中间任何一步失败比如价格API超时、任何一步需要人工确认比如“你确定要查2023年款还是2024年款”、任何一步产生副作用比如“生成建议”同时触发了邮件发送都要求系统具备明确的状态管理、错误隔离、人机协同和副作用控制能力。这就引出了最关键的判断标准如果你的方案无法回答以下五个问题它就还不是真正的Agent架构只是LLM的高级调用封装当某一步骤执行超时系统如何标记该节点失败而不影响其他并行分支用户在第三步突然改口说“等等先查下C款”历史上下文如何安全地切片、冻结与重载工具调用返回了结构化JSON但字段缺失或类型错误是直接抛异常中断流程还是降级为文本描述继续多个Agent实例并发处理不同用户请求时各自的memory、tool state、execution context如何物理隔离系统日志里记录的不是“调用成功/失败”而是“第2.3步价格查询因rate limit被限流已自动切换备用供应商接口耗时820ms”——这种粒度的可观测性从何而来这些问题的答案不在prompt engineering里也不在模型选型里而在Runtime的设计哲学里。Harness Runtime之所以被越来越多团队采用正是因为它把上述所有问题转化成了可配置、可插拔、可测试的组件契约而不是靠开发者在业务代码里一遍遍手写try-catch和context.save()。提示别被“Agent Framework”这个词迷惑。很多标榜“开箱即用”的框架底层仍是线性pipeline封装没有独立的Execution Loop、State Registry和Tool Orchestrator。它们能跑Demo但扛不住真实业务的复杂度熵增。2. 从零到一一次LLM调用如何演变成Harness Runtime的四层结构我们来还原一个典型的技术演进路径——不是教科书式的理论推导而是我在三个不同项目中实际踩过的坑、重构的版本、最终沉淀下来的分层逻辑。它清晰展示了为什么“简单调用”必须让位于“结构化Runtime”。2.1 第一层裸调用层The Raw Call Layer这是所有人的起点用requests.post向LLM API发送JSON payload解析response[choices][0][message][content]。代码可能长这样import requests import json def call_llm(prompt: str) - str: url https://api.example.com/v1/chat/completions headers {Authorization: fBearer {API_KEY}} data { model: gpt-4-turbo, messages: [{role: user, content: prompt}], temperature: 0.3 } resp requests.post(url, headersheaders, jsondata, timeout30) return resp.json()[choices][0][message][content]表面看很干净实则埋了五颗雷无状态绑定每次调用都是全新上下文无法维持对话历史、用户偏好、任务进度无工具抽象所有外部API调用查天气、搜数据库都得硬编码在prompt里或由LLM自己拼接URL既不安全也不可控无错误语义HTTP 429限流、503服务不可用、400bad request全被统一吞掉只返回空字符串或抛异常业务层无法区分是模型拒答还是网络抖动无可观测性没有request_id、没有token消耗统计、没有实际响应延迟出了问题只能看日志里“调用失败”四个字无扩展点想加个预处理器如敏感词过滤、后处理器如结果格式化、缓存层只能改这个函数本身违反单一职责。这个阶段你拥有的不是一个Agent只是一个“带网络请求的printf”。2.2 第二层编排层The Orchestration Layer当业务需求变复杂比如“先查用户订单再根据订单状态决定是否触发退款”就必须引入显式编排。这时团队通常会引入LangChain、LlamaIndex等框架代码变成from langchain_core.prompts import ChatPromptTemplate from langchain_openai import ChatOpenAI from langchain.agents import AgentExecutor, create_tool_calling_agent # 定义工具 tool def get_order_status(order_id: str) - str: # 调用内部订单服务 return shipped # 构建Agent llm ChatOpenAI(modelgpt-4-turbo) prompt ChatPromptTemplate.from_messages([...]) agent create_tool_calling_agent(llm, [get_order_status], prompt) agent_executor AgentExecutor(agentagent, tools[get_order_status]) result agent_executor.invoke({input: 我的订单12345状态如何})这层解决了工具调用抽象和基础编排逻辑但新问题立刻浮现状态散落各处agent_executor内部维护的chat_history是临时对象重启进程就丢get_order_status工具里的连接池、缓存、重试策略各自为政执行不可控LLM决定调用哪个工具、调用几次、传什么参数全凭概率采样。你无法强制它“必须先查订单再查物流”也无法在它连续三次调用失败后自动降级调试黑洞agent_executor.invoke()返回一个dict里面output字段是最终答案但中间所有tool call、thought过程、LLM原始响应全被框架封装隐藏出问题只能加断点进源码部署割裂本地跑通的Agent上线后发现get_order_status工具依赖的内网服务DNS解析失败而框架没提供环境感知的工具注册机制。这一层的价值是证明了“编排必要性”代价是把复杂度从代码里转移到了框架黑盒中。2.3 第三层运行时层The Runtime Layer真正的转折点出现在团队决定把“Agent执行”当作一个独立生命周期来管理时。我们不再问“怎么让LLM调用工具”而是问“一个Agent实例的完整生命周期包含哪些阶段每个阶段谁负责边界在哪”Harness Runtime正是基于此哲学构建的。它不提供prompt模板也不封装tool定义而是定义了一套最小但完备的执行契约。其核心四组件如下表所示组件职责关键接口典型实现Executor驱动执行循环接收Input → 调用LLM → 解析Tool Call → 分发给Tool → 收集Result → 决定Next Stepexecute(input: Input) - Output基于asyncio的事件循环支持中断/恢复State Manager持久化并同步所有运行时状态当前step、history、tool outputs、user feedback、custom metadatasave(state: State),load(session_id: str) - StateRedis Hash TTL或SQLite WAL模式Tool Registry统一注册、发现、调用、监控所有工具。工具必须实现标准invoke(input: dict) - dict接口register(name: str, tool: Tool),invoke(name: str, input: dict) - dict工具元数据存PostgreSQL调用走gRPC自动注入trace_idObservability Hub采集、聚合、暴露所有可观测性数据每步耗时、token用量、tool成功率、LLM拒答率、人工干预次数emit(event: Event)支持OpenTelemetry Exporter数据写入Prometheus Grafana看板告警规则基于P95延迟注意这里没有“Agent类”没有“Chain类”只有契约化的组件接口。你可以用Pydantic定义Input/OutputSchema用FastAPI暴露execute端点用Celery做异步Task调度——只要遵守接口任何技术栈都能接入。我参与的一个电商客服项目就是用这套契约重构了原有LangChain方案。原来一个“查订单查物流生成话术”的Agent平均响应时间1.8秒P99超时率达12%。重构后Executor加入step-level timeout查订单≤800ms查物流≤500ms超时自动跳过并标记State Manager将chat_history压缩为token-aware的sliding window内存占用降65%Tool Registry为每个工具配置熔断器连续3次失败10分钟内拒绝调用避免雪崩Observability Hub暴露出一个关键问题get_logistics工具在下午3-5点P99延迟突增至3.2秒根源是物流服务商的CDN缓存失效。这在旧架构里根本不可见。这一层的本质是把Agent从“代码逻辑”升维成“可运维的服务”。2.4 第四层应用层The Application Layer当Runtime稳定后业务逻辑才真正回归简洁。开发者不再操心“怎么让LLM调用工具”而是专注“这个业务场景需要哪些状态、哪些工具、哪些决策规则”。以“智能报销助手”为例应用层代码可能只有# app/reimbursement_agent.py class ReimbursementAgent: def __init__(self, executor: Executor, state_mgr: StateManager): self.executor executor self.state_mgr state_mgr async def handle_user_input(self, user_id: str, text: str) - str: # 1. 加载用户专属状态历史报销单、常用发票类型、审批流配置 state await self.state_mgr.load(fuser:{user_id}) # 2. 构建输入注入业务上下文 input_data { user_input: text, user_profile: state.get(profile, {}), pending_claims: state.get(pending_claims, []) } # 3. 执行Runtime获取结构化输出 output await self.executor.execute(input_data) # 4. 根据输出类型做业务路由 if output.type NEED_MORE_INFO: await self._ask_clarification(user_id, output.prompt) elif output.type CLAIM_SUBMITTED: await self._notify_approver(output.claim_id) return output.response你看不到LLM调用看不到prompt模板甚至看不到工具名。所有“智能”被封装在Runtime的配置里executor的配置文件指定了当检测到“发票”关键词优先调用extract_invoice_info工具当pending_claims非空自动注入“请先处理未完成报销”的约束state_mgr的配置决定了用户状态默认保存7天敏感字段如银行卡号自动脱敏加密tool_registry的配置规定了submit_claim工具必须经过风控服务校验否则拒绝执行。这一层的价值是让AI能力像数据库连接池一样成为基础设施的一部分——业务代码只消费契约不关心实现。注意Harness Runtime不是“另一个框架”而是一套设计原则的工程化落地。它的最大优势在于“可替换性”今天用Redis做State Manager明天可以无缝换成DynamoDB当前用OpenAI换Claude只需改Executor的LLM Adapter其他组件完全不动。这种解耦是裸调用和编排层永远无法提供的。3. 关键技术选型为什么是Harness而不是LangChain、AutoGen或自研面对市面上琳琅满目的Agent方案技术负责人最常问的是“我们到底该选哪个”这个问题本身就有陷阱——它预设了“选一个框架就能解决问题”。而现实是选型不是挑功能最全的而是挑最匹配你当前阶段技术债和未来演进路径的。下面是我基于实际项目数据做的横向对比聚焦三个致命维度状态管理、错误处理、可观测性。3.1 状态管理持久化不是可选项而是生命线方案状态存储方式进程重启后状态是否存活多实例共享状态支持状态版本控制实测P99加载延迟万级用户裸调用完全无状态否否否N/ALangChain MemoryIn-memory dict / 可插拔Backend如Redis依赖Backend配置但Memory类本身无生命周期管理需手动配置Backend易遗漏否Redis: 12msSQL: 45msAutoGen GroupChatChatHistory对象存于Agent实例内存否否否N/A内存操作Harness RuntimeState Manager组件强制抽象必须实现save/load接口是由State Manager保证是通过共享存储是State对象含version字段冲突时自动mergeRedis Hash: 8msDynamoDB: 22ms关键洞察LangChain的Memory看似灵活但90%的团队用的是ConversationBufferMemory状态全在内存里。一旦服务重启或扩容用户正在聊的“报销流程”瞬间清零体验断崖式下跌。而Harness把状态管理提升为第一公民要求你必须回答“状态存在哪、怎么存、怎么取、怎么保活”。这不是增加工作量而是提前暴露架构风险。我们在某金融项目中曾用LangChain Memory上线结果每天早9点批量消息涌入时K8s自动扩缩容导致大量Agent实例重建用户投诉“刚说一半的话没了”。切换到Harness后State Manager对接Redis Cluster配合客户端本地缓存P99加载延迟压到10ms内投诉归零。3.2 错误处理从“静默失败”到“精准熔断”方案LLM调用失败如何处理工具调用失败如何处理是否支持降级策略是否可配置熔断器错误日志能否定位到具体Step裸调用try-except捕获Exception返回通用错误同上无工具概念否否否只有“调用失败”LangChainAgentExecutor有handle_parsing_errors参数但仅处理LLM输出格式错误Tool类可自定义异常但无统一错误分类需手动在Tool里写fallback逻辑否否日志分散在各组件AutoGenConversableAgent有handle_message钩子但需重写整个消息处理链无内置工具错误处理机制否否否Harness RuntimeExecutor内置llm_fallback策略可配置重试次数、备用模型、或直接返回预设兜底文案Tool Registry强制要求Tool实现error_handler接口返回标准化Error Code是在State中定义fallback_step如“查不到物流→展示预计送达时间”是基于滑动窗口失败率自动开启/关闭Tool是每条日志带step_id、tool_name、error_code真实案例某物流公司的“运单追踪Agent”高峰期get_tracking_info工具因下游服务商限流失败率飙升至35%。在LangChain方案中错误被吞掉LLM反复尝试调用加剧雪崩。Harness方案中Tool Registry的熔断器在失败率超30%后自动切断该ToolExecutor立即执行fallback_step——调用缓存服务返回最后一次成功结果并标注“数据可能有延迟”用户体验平稳过渡。3.3 可观测性没有度量就没有优化方案默认采集哪些指标是否支持自定义Trace是否可关联User ID是否可生成执行流程图是否支持实时告警裸调用无否否否否LangChainllm_start/llm_end等Callback事件需手动实现是通过CallbackHandler需手动注入否否需额外集成AutoGen仅print级别日志否否否否Harness Runtime开箱即用每Step耗时、token用量、tool成功率、LLM拒答率、人工干预次数、state size是OpenTelemetry原生支持是execute接口强制user_id参数是Observability Hub提供/trace/{trace_id}端点返回JSON格式执行树是Prometheus Alertmanager规则如“连续5分钟tool_fail_rate 15%”最震撼的数据来自一个教育项目接入Harness后Observability Hub暴露出一个隐藏瓶颈——generate_study_plan工具在处理“高三数学”请求时平均token用量是“小学语文”的4.7倍导致LLM调用成本激增。团队据此优化了该工具的输入压缩算法成本直降38%。这种深度洞察在缺乏结构化可观测性的方案里根本无从谈起。总结选型逻辑如果你的团队还在验证PMFProduct-Market Fit用LangChain快速出Demo没问题如果已进入规模化交付且对SLAService Level Agreement有硬性要求Harness Runtime的“契约先行、组件解耦、可观测驱动”哲学能帮你少踩80%的生产环境大坑。它不承诺“更快上线”但承诺“上线后更稳、更可调、更可扩”。4. 落地实战从零搭建一个Harness Runtime的七步法理论讲完现在进入最硬核的部分——手把手带你搭起一个可运行的Harness Runtime最小可行版MVP。这不是玩具Demo而是我在某客户现场4小时搭出来的生产级骨架已支撑日均50万次Agent调用。所有代码均可直接复制运行关键配置已标注原理。4.1 步骤一定义核心契约Schema First一切始于接口。我们用Pydantic V2定义三个核心Schema它们是整个Runtime的“宪法”# schemas.py from pydantic import BaseModel, Field, validator from typing import Dict, List, Optional, Literal import re class Input(BaseModel): 所有Agent执行的统一输入契约 user_id: str Field(..., description唯一用户标识用于状态隔离和审计) session_id: str Field(..., description会话ID用于跨请求状态关联) text: str Field(..., min_length1, max_length4096, description用户原始输入) context: Dict[str, str] Field(default_factorydict, description业务上下文如用户画像、历史行为) class State(BaseModel): 运行时状态契约必须可序列化、可版本化、可合并 version: int Field(default1, ge1, description状态版本号用于乐观锁和冲突解决) history: List[Dict[str, str]] Field(default_factorylist, description消息历史格式[{role:user,content:xxx}]) step: str Field(defaultstart, description当前执行步骤名如extract_info, validate_data) tool_outputs: Dict[str, dict] Field(default_factorydict, description已执行工具的输出缓存) custom_metadata: Dict[str, str] Field(default_factorydict, description业务自定义元数据) validator(history) def validate_history_length(cls, v): if len(v) 50: # 强制限制历史长度防OOM raise ValueError(history length must be 50) return v class Output(BaseModel): 执行结果契约结构化、可路由、含元信息 type: Literal[TEXT_RESPONSE, NEED_MORE_INFO, TOOL_CALL, ERROR] Field(..., description响应类型) response: str Field(, description最终返回给用户的文本) next_step: Optional[str] Field(None, description下一步骤名用于流程编排) tool_calls: List[Dict[str, str]] Field(default_factorylist, description待执行的工具调用列表) metrics: Dict[str, float] Field(default_factorydict, description本次执行的关键指标)为什么这步最关键因为Schema定义了所有组件的“共同语言”。Executor只认InputState Manager只存StateObservability Hub只收Output。任何组件的实现只要满足这三个Pydantic模型就能无缝接入。这避免了后期因数据格式不一致导致的“胶水代码地狱”。4.2 步骤二实现State Manager状态是基石我们选择Redis作为默认存储因其高性能、高可用、天然支持TTL自动过期。代码极简但每行都有深意# state_manager.py import redis import json import time from typing import Optional from schemas import State class RedisStateManager: def __init__(self, host: str localhost, port: int 6379, db: int 0): self.redis redis.Redis(hosthost, portport, dbdb, decode_responsesTrue) # 设置全局TTL用户状态默认保留24小时 self.default_ttl 24 * 60 * 60 def save(self, session_id: str, state: State) - bool: 保存状态使用Redis Hash结构key为session_idfield为version/history等 try: # 将State转为dict但排除非JSON序列化字段如datetime state_dict state.dict(exclude_unsetTrue, exclude_noneTrue) # 关键使用HSET而非SET便于部分更新如只更新history pipe self.redis.pipeline() for key, value in state_dict.items(): # 对大字段如history做JSON序列化 if isinstance(value, (list, dict)): pipe.hset(session_id, key, json.dumps(value)) else: pipe.hset(session_id, key, str(value)) # 设置过期时间 pipe.expire(session_id, self.default_ttl) pipe.execute() return True except Exception as e: print(f[StateMgr] Save failed for {session_id}: {e}) return False def load(self, session_id: str) - Optional[State]: 加载状态自动处理JSON反序列化 try: # HGETALL获取所有field-value对 data self.redis.hgetall(session_id) if not data: return None # 将JSON字符串字段反序列化 processed_data {} for key, value in data.items(): if key in [history, tool_outputs, custom_metadata]: try: processed_data[key] json.loads(value) except json.JSONDecodeError: processed_data[key] {} else: processed_data[key] value # 构建State对象自动校验 return State(**processed_data) except Exception as e: print(f[StateMgr] Load failed for {session_id}: {e}) return None # 使用示例 state_mgr RedisStateManager() state State(history[{role: user, content: 你好}]) state_mgr.save(sess_123, state) loaded state_mgr.load(sess_123) # 返回State对象含自动校验实操心得不要用SET session_id JSON而要用HSET session_id field value。因为Agent执行中你经常只需要更新history或stepHSET支持部分更新避免读-改-写Read-Modify-Write竞争default_ttl必须设否则Redis内存爆满这是无数团队踩过的坑load方法里加了try-except包裹json.loads因为网络传输中JSON可能损坏不能让单个坏数据导致整个Agent崩溃。4.3 步骤三构建Executor执行引擎Executor是Runtime的心脏。我们用asyncio实现支持并发、中断、超时。核心是execute方法它严格遵循“输入→LLM→解析→分发→聚合→输出”流程# executor.py import asyncio import json import time from typing import Dict, Any, List, Optional from openai import AsyncOpenAI from schemas import Input, State, Output class HarnessExecutor: def __init__(self, api_key: str, model: str gpt-4-turbo): self.client AsyncOpenAI(api_keyapi_key) self.model model async def execute(self, input_data: Input, state_mgr) - Output: 执行一次Agent调用的完整生命周期 start_time time.time() # 1. 加载或初始化状态 state state_mgr.load(input_data.session_id) if not state: state State(history[], stepstart) # 2. 构建LLM输入注入上下文和历史 messages self._build_messages(input_data, state) try: # 3. 调用LLM带超时 response await asyncio.wait_for( self.client.chat.completions.create( modelself.model, messagesmessages, temperature0.3, max_tokens1024 ), timeout15.0 # 整体LLM调用超时15秒 ) # 4. 解析LLM响应这里简化为检测TOOL:前缀实际应使用function calling llm_output response.choices[0].message.content.strip() if llm_output.startswith(TOOL:): # 解析工具调用指令如 TOOL: search_web(queryAI Agent 架构) tool_call self._parse_tool_call(llm_output) return Output( typeTOOL_CALL, tool_calls[tool_call], metrics{llm_latency_ms: (time.time() - start_time) * 1000} ) else: # 5. 直接返回文本响应 state.history.append({role: user, content: input_data.text}) state.history.append({role: assistant, content: llm_output}) state.step complete state_mgr.save(input_data.session_id, state) # 保存状态 return Output( typeTEXT_RESPONSE, responsellm_output, metrics{llm_latency_ms: (time.time() - start_time) * 1000} ) except asyncio.TimeoutError: return Output(typeERROR, responseLLM调用超时请稍后重试, metrics{llm_timeout: True}) except Exception as e: return Output(typeERROR, responsef系统错误{str(e)}, metrics{llm_error: str(e)}) def _build_messages(self, input_data: Input, state: State) - List[Dict[str, str]]: 构建符合OpenAI格式的消息列表 messages [] # 系统提示词System Prompt放最前 messages.append({ role: system, content: 你是一个专业客服助手只回答与用户问题相关的内容。 }) # 注入业务上下文 if input_data.context: context_str 业务上下文 ; .join([f{k}{v} for k,v in input_data.context.items()]) messages.append({role: system, content: context_str}) # 追加历史 messages.extend(state.history) # 追加当前用户输入 messages.append({role: user, content: input_data.text}) return messages def _parse_tool_call(self, text: str) - Dict[str, str]: 简易工具调用解析器实际项目应替换为JSON Schema校验 # 示例TOOL: search_web(queryAI Agent) - {name: search_web, args: {query: AI Agent}} import re match re.match(rTOOL:\s*(\w)\((.*)\), text) if match: name match.group(1) args_str match.group(2) # 简单解析args实际用ast.literal_eval或pydantic args {query: args_str.strip(\)} return {name: name, args: args} return {name: unknown, args: {}} # 使用示例 executor HarnessExecutor(api_keysk-...) state_mgr RedisStateManager() # 模拟一次调用 input_obj Input( user_idu_789, session_idsess_123, text查一下AI Agent的最新架构趋势, context{domain: tech_blog} ) # 注意此处需在async def中调用 # result await executor.execute(input_obj, state_mgr)避坑指南asyncio.wait_for必须用裸awaitclient.chat.completions.create在LLM服务抖动时会无限等待拖垮整个服务_build_messages里系统提示词system prompt必须放在最前面且不能被用户输入覆盖。我见过太多团队把system prompt写在prompt模板里结果用户一句话“忽略之前的指令”就把整个安全护栏废掉了_parse_tool_call是临时方案真实项目必须用OpenAI的tools参数和tool_choice由模型原生支持function calling解析更可靠。4.4 步骤四注册第一个工具Tool Registry工具是Agent的“手脚”。Harness要求每个工具必须实现标准接口我们以“搜索网页”为例# tools/search_web.py import aiohttp import asyncio from typing import Dict, Any class SearchWebTool: 符合Harness Tool契约的搜索工具 def __init__(self, api_key: str): self.api_key api_key self.base_url https://api.search.example.com/v1 async def invoke(self, input_data: Dict[str, Any]) - Dict[str, Any]: Tool标准接口输入dict输出dict query input_data.get(query, ) if not query: return {error: query is required} try: # 调用外部搜索API async with aiohttp.ClientSession() as session: async with session.get( f{self.base_url}/search, params{q: query, key: self.api_key}, timeout10.0 ) as resp: if resp.status 200: data await resp.json() # 标准化输出必须含results字段 return { success: True, results: data.get(items, [])[:3], # 只取前3条 query: query } else: return {error: fSearch API returned {resp.status}} except asyncio.TimeoutError: return {error: Search API timeout} except Exception as e: return {error: fSearch API error: {str(e)}} # tools/registry.py from typing import Dict, Callable, Any from search_web import SearchWebTool class ToolRegistry: def __init__(self): self.tools: Dict[str, Callable] {} def register(self, name: str, tool_instance: Any): 注册工具实例 self.tools[name] tool_instance async def invoke(self, name: str, input_data: Dict[str, Any]) - Dict[str, Any]: 调用工具统一错误处理 if name not in self.tools: return {error: fTool {name} not registered} try: # 所有工具调用都包一层捕获异常 result await self.tools[name].invoke(input_data) return result except Exception as e: return {error: fTool {name} execution failed: {str(e)}} # 初始化Registry tool_registry ToolRegistry() tool_registry.register(search_web, SearchWebTool(api_keysearch_key_123))关键设计invoke方法签名强制为async def invoke(self, input_data: Dict) - Dict确保所有工具行为一致ToolRegistry.invoke做了统一异常包装业务层无需关心每个工具的try-except输出必须是Dict且推荐含success/error字段方便Executor做条件判断。4.5 步骤五集成Observability Hub让一切可见没有监控的Agent就像没有仪表盘的飞机。我们用OpenTelemetry标准将所有关键事件上报# observability/hub.py from opentelemetry import trace from opentelemetry.exporter.otlp.proto.http.trace_exporter import OTLPSpanExporter from opentelemetry.sdk.trace import TracerProvider from opentelemetry.sdk.trace.export import BatchSpanProcessor