1. 这不是概念堆砌而是Agent落地时每天要面对的真实战场“Harness、Loop、GraphAgent 工程的三层架构与生产实践全解析”——这个标题乍看像学术论文但如果你正在真实推进一个能跑通业务闭环的Agent项目比如让客服Agent自动处理80%的退换货请求、让运维Agent在凌晨三点自主定位并回滚异常服务、或者让投研Agent持续抓取财报数据交叉验证生成风险提示报告那你立刻会意识到这三层不是理论分层而是你每天调试日志、优化延迟、排查超时、说服产品接受“不能100%准确”的时候背后真正起作用的三道技术防线。我带团队做过6个从0到1上线的Agent系统覆盖金融风控、工业设备预测性维护、跨境电商多语言客服三个完全不同的领域。最深的体会是90%的失败不是卡在大模型能力上而是卡在这三层之间的缝隙里——Harness层把Prompt塞进去却收不到结构化输出Loop层想重试三次结果第二次就触发了风控熔断Graph层设计了5个节点协同但其中两个节点永远等不到上游传来的context。这些不是PPT里的箭头连接而是线上告警、用户投诉、SLA违约背后的具体代码行和配置参数。核心关键词“Harness、Loop、Graph”必须放在真实工程语境里理解Harness是Agent的“躯干”它不负责思考但决定思考能否发生——它封装了模型调用、输入清洗、输出解析、错误兜底、成本控制。没有健壮的Harness再聪明的Agent也像没装操作系统的芯片。Loop是Agent的“神经反射弧”它不定义目标但决定目标能否达成——它管理状态流转、重试策略、人工干预点、超时熔断、结果校验。没有可控的LoopAgent要么死循环要么一击即溃。Graph是Agent的“大脑皮层”它不执行计算但决定计算如何组织——它编排节点依赖、传递上下文、聚合多源结果、处理分支逻辑。没有清晰的GraphAgent就是一堆孤立函数无法应对复杂任务。这篇文章不讲LLM原理不对比各家模型API只聚焦一件事当你手上有业务需求、有可用模型、有开发资源如何用Harness/Loop/Graph这三层把Agent从Demo变成可监控、可扩缩、可迭代的生产服务。适合两类人一是正被“Agent怎么落地”困扰的工程师二是需要评估Agent项目可行性的技术负责人。下面所有内容都来自我们踩过的坑、压测的数据、线上监控截图和深夜改完的配置文件。2. Harness层不是简单封装API而是构建Agent的“呼吸系统”2.1 Harness的本质是协议转换器不是HTTP客户端很多团队第一步就错了把Harness当成一个“调用大模型API的工具类”。结果写了一堆model.invoke(prompt)然后发现输出格式不稳定、token超限没预警、错误类型混乱、成本无法分摊。这就像给汽车装了个能点火的按钮却没配油路、电路、冷却系统。Harness真正的职责是完成四重协议转换业务协议 → 模型协议把订单ID、用户画像、历史对话等业务字段按模型要求的JSON Schema或XML结构组装成Prompt模型协议 → 业务协议把模型返回的自由文本用Schema约束正则校验LLM后处理强制转成{ action: REFUND, amount: 299.0, reason: damaged }这样的确定性结构成本协议 → 财务协议把input_tokens: 1247, output_tokens: 89实时换算成人民币金额考虑不同模型、不同region的单价并打标到业务维度如“客服退换货场景”错误协议 → 运维协议把429 Rate Limit、503 Service Unavailable、output parsing failed等17类错误映射成统一的ERROR_CODE如HARNESS_RATE_LIMIT_EXCEEDED并携带trace_id、model_name、prompt_length等上下文直送监控系统。提示我们曾因忽略第2点付出代价。某次电商大促模型返回的退款金额偶尔带单位“¥”导致下游财务系统解析失败。后来在Harness层加了强制数值校验if not isinstance(output[amount], (int, float)) or output[amount] 0: raise ParsingError(amount must be positive number)。这行代码上线后该错误归零。2.2 实战中的Harness设计以“客服退换货Agent”为例我们为某电商平台设计的Harness核心模块如下非伪代码是真实生产级结构class RefundHarness: def __init__(self, model_client: ModelClient, config: HarnessConfig): self.model_client model_client # 封装OpenAI/DeepSeek/自研模型SDK self.config config # 包含timeout_ms3000, max_retries2等 self.parser JSONSchemaParser(schemaREFUND_SCHEMA) # 预编译Schema self.cost_tracker CostTracker() # 实时计算token成本 def invoke(self, user_input: str, context: dict) - RefundDecision: # Step 1: 构建Prompt业务协议→模型协议 prompt self._build_prompt(user_input, context) # Step 2: 调用模型带熔断和重试 try: raw_response self.model_client.invoke( promptprompt, temperatureself.config.temperature, max_tokensself.config.max_output_tokens, timeoutself.config.timeout_ms ) except ModelTimeoutError as e: raise HarnessError(MODEL_TIMEOUT, causee, contextcontext) except ModelRateLimitError as e: raise HarnessError(RATE_LIMIT_EXCEEDED, causee, contextcontext) # Step 3: 解析输出模型协议→业务协议 try: parsed self.parser.parse(raw_response) except ValidationError as e: # 关键对解析失败做分级处理 if self._is_structural_failure(raw_response): # 如返回纯文本无JSON raise HarnessError(OUTPUT_FORMAT_INVALID, causee, contextcontext) else: # 如金额超范围尝试LLM后处理修复 repaired self._repair_with_llm(raw_response, REFUND_SCHEMA) if repaired: parsed repaired else: raise HarnessError(OUTPUT_SEMANTIC_INVALID, causee, contextcontext) # Step 4: 成本核算与审计 cost self.cost_tracker.calculate( input_tokenslen(prompt), output_tokenslen(raw_response), model_nameself.model_client.model_name ) audit_log AuditLog( trace_idcontext.get(trace_id), harness_versionv2.3.1, costcost, input_hashhashlib.md5(prompt.encode()).hexdigest() ) audit_log.save() return RefundDecision(**parsed)这个Harness的关键设计选择及其理由预编译JSON Schema而非运行时解析REFUND_SCHEMA是Pydantic V2的BaseModel在服务启动时编译成Cython加速的validator。实测比jsonschema.validate()快4.7倍且内存占用降低62%。因为客服场景QPS峰值达1200每次解析耗时必须5ms。解析失败的分级处理不是简单抛错。结构性失败无JSON立即上报语义性失败金额负数先用轻量LLM修复调用更小、更快的模型重写关键字段。我们在压测中发现23%的解析失败属于后者修复成功率89%避免了30%的无效人工介入。审计日志带input_hash不是记录原始Prompt太长且含敏感信息而是存MD5哈希。当线上出现争议时可通过hash快速检索原始日志既满足审计要求又规避隐私风险。2.3 Harness层避坑指南那些文档里不会写的细节问题现象根本原因我们的解决方案效果同一Prompt多次调用输出差异大温度值设为1.0模型随机性失控在Harness层强制temperature0.3对确定性任务如分类、提取禁用top_p采样输出一致性从72%提升至99.8%Token计费远超预期Prompt中混入大量空格、换行、注释文本在_build_prompt()中增加clean_prompt()移除多余空白、压缩JSON key、替换长描述为短code如customer_complaint_reason→ccr平均输入token减少38%成本下降29%模型返回空字符串或乱码网络抖动导致HTTP响应体截断在model_client.invoke()中增加响应完整性校验if len(raw_response) 10 or raw_response.strip() in [, null, {}]:截断错误捕获率100%避免下游空指针异常无法区分是模型问题还是业务逻辑问题错误日志只记录model.invoke failed所有HarnessError携带error_sourcemodel或parser或network标签并在监控大盘按source分组故障定位时间从平均47分钟缩短至8分钟注意不要在Harness层做业务规则判断。曾有团队在Harness里加了“如果用户说‘我要投诉’直接转人工”这违反了分层原则——Harness只管“能不能调通”Loop层才管“要不要转人工”。混淆会导致Harness臃肿、难以复用、测试爆炸。3. Loop层不是while True而是Agent的“决策中枢神经系统”3.1 Loop的本质是状态机不是重试逻辑把Loop理解为“调用失败就retry 3次”是最大的认知误区。Loop层要解决的核心问题是当Agent面对一个开放性任务如“帮用户解决退货问题”时如何在不确定的环境中基于有限反馈逐步收敛到可交付结果。这需要一套完整的状态机设计State状态不是简单的RUNNING/FAILED/DONE而是WAITING_FOR_USER_CONFIRMATION、RETRYING_WITH_ENHANCED_CONTEXT、AWAITING_EXTERNAL_API_RESULT、NEEDS_HUMAN_APPROVAL等12种业务态Transition流转每个状态都有明确的进入条件Entry Condition和退出动作Exit Action。例如进入NEEDS_HUMAN_APPROVAL态的条件是“退款金额500元且用户信用分600”退出动作是“将工单推送到CRM系统设置SLA为2小时”Guard守卫防止非法流转。如WAITING_FOR_USER_CONFIRMATION态下若用户30秒未回复自动转入RETRYING_WITH_ENHANCED_CONTEXT态而不是直接失败Side Effect副作用状态变更必须触发可观测行为。进入AWAITING_EXTERNAL_API_RESULT态时必须记录API调用详情、启动超时定时器、向监控系统发loop_state_change事件。我们为工业设备预测性维护Agent设计的Loop状态机简化版[INIT] ↓ (start_maintenance_task) [ANALYZE_SENSOR_DATA] → [DATA_INSUFFICIENT]? → [REQUEST_MORE_DATA] → [WAIT_FOR_DATA] → [ANALYZE_SENSOR_DATA] ↓ (anomaly_confirmed) ↑ (timeout_30s) [PLAN_REPAIR_ACTION] → [ACTION_UNCERTAIN]? → [CONSULT_KNOWLEDGE_BASE] → [PLAN_REPAIR_ACTION] ↓ (plan_approved) ↑ (no_matching_knowledge) [EXECUTE_REPAIR] → [EXECUTION_FAILED]? → [ADJUST_PLAN] → [EXECUTE_REPAIR] ↓ (success) ↑ (max_retry_exceeded) [DONE] ↓ (human_required) [HUMAN_INTERVENTION_REQUIRED]这个Loop的价值在于所有决策点都可配置、可审计、可降级。例如当知识库查询失败时不是整个Agent崩掉而是降级到ADJUST_PLAN态用规则引擎生成备选方案。3.2 Loop层的核心实现以“跨平台客服Agent”为例该Agent需在微信、APP、网页三端协同处理用户咨询。Loop层代码框架如下class CrossPlatformLoop: def __init__(self, harness: Harness, graph: GraphEngine): self.harness harness self.graph graph self.state_machine StateMachine(definitionLOOP_DEFINITION) # 状态机定义 def run(self, initial_context: dict) - LoopResult: state self.state_machine.initial_state context initial_context.copy() loop_count 0 max_loop 15 # 防死循环硬限制 while state ! DONE and loop_count max_loop: loop_count 1 # Step 1: 执行当前状态的Action try: action_result self._execute_state_action(state, context) except LoopError as e: # Loop层自己的错误如状态机配置错误非Harness错误 return LoopResult(failedTrue, error_codee.code, contextcontext) # Step 2: 更新Context关键Loop的生命线 context self._update_context(context, action_result) # Step 3: 根据Action结果和Guard条件决定下一个State next_state self.state_machine.transition( current_statestate, eventaction_result.event, # 如USER_CONFIRMED, API_TIMEOUT contextcontext, guard_params{ # Guard计算所需参数 elapsed_time: time.time() - context.get(start_time, 0), retry_count: context.get(retry_count, 0) } ) # Step 4: 执行State Exit Action如发消息、写DB self._execute_exit_action(state, next_state, context) state next_state if state DONE: return LoopResult(successTrue, final_contextcontext) else: return LoopResult(failedTrue, error_codeLOOP_MAX_ITERATION_EXCEEDED, contextcontext) def _execute_state_action(self, state: str, context: dict) - ActionResult: 每个State对应一个Action函数解耦业务逻辑 action_map { ANALYZE_QUERY: self._analyze_user_query, FETCH_ORDER_HISTORY: self._fetch_order_history, GENERATE_REFUND_PLAN: self._generate_refund_plan, WAIT_FOR_USER_CONFIRMATION: self._wait_for_confirmation, } return action_map[state](context)关键设计点解析Context是唯一真相源所有状态共享同一个context字典它存储user_id,last_message_time,retry_count,pending_api_calls等全局状态。_update_context()不是简单merge而是用deep_update()确保嵌套结构正确合并避免context[order][status]被意外覆盖。Event驱动而非轮询action_result.event是状态流转的唯一依据。例如_wait_for_confirmation()返回eventUSER_TIMEOUT触发Guard检查elapsed_time 30从而跳转到重试态。这比time.sleep(30)更可靠支持异步回调。Guard参数隔离Guard计算所需的elapsed_time、retry_count等通过guard_params显式传入避免Action函数污染Context。这使Guard逻辑可单元测试且不依赖具体实现。3.3 Loop层实战经验让Agent学会“适时放弃”Loop层最难的部分不是让它继续而是让它优雅放弃。我们总结出三条铁律第一设置“放弃阈值”必须量化。不能写“如果用户不回复就放弃”而要定义时间阈值WAIT_FOR_USER_CONFIRMATION态下last_message_time 300s now()成本阈值单次Loop消耗token超过5000或累计成本超2.5确定性阈值harness.invoke()连续3次返回confidence_score 0.6Harness层需返回置信度。第二放弃必须有“移交路径”。放弃不是结束而是转交。我们的标准移交路径生成结构化摘要含时间线、已尝试步骤、失败原因推送至指定队列如human_handoff_queue向用户发送移交确认消息“您的问题已转交高级专员将在2小时内回复”记录移交ID供后续追溯。第三放弃决策本身要可审计。我们在Loop层埋点if should_abandon: audit_event { event: LOOP_ABANDONED, abandon_reason: TIMEOUT_IN_WAIT_FOR_CONFIRMATION, context_snapshot: {k: v for k, v in context.items() if k in [user_id, session_id, loop_step_count]}, decision_trace: [guard_elapsed_time302s 300s, retry_count0] } logger.audit(audit_event) # 发送到专用审计日志流这让我们能回答“为什么这个Case被转人工”——不是靠猜而是查日志。4. Graph层不是画流程图而是Agent的“认知拓扑结构”4.1 Graph的本质是上下文编织器不是DAG调度器很多团队用Airflow或Prefect来编排Agent节点结果发现节点间传递的是{data: {...}}这种万能字典导致下游节点总要if order_id in data做防御性编程图形界面拖拽出来的DAG无法表达“如果节点A失败则跳过B直接执行C但C的输入要从A的原始输入重构”这种复杂依赖当需要动态增删节点如根据用户等级插入风控校验节点时DAG配置要重启服务。Graph层真正的挑战是在动态、异构、部分失败的环境中保证上下文Context的完整性、一致性、可追溯性。它不是静态的执行计划而是运行时的上下文拓扑图。我们定义Graph的三个核心要素Node节点不是函数而是NodeDef对象包含id,type(LLM/Rule/API),input_schema,output_schema,fallback_node_idEdge边不是A→B而是A.output_field_x → B.input_field_y的精确映射支持transform函数如str.upper()Context Graph上下文图运行时动态构建的图每个节点执行后其输出被注入到全局Context的指定路径如/nodes/analyze_query/result边的映射关系决定哪些路径被读取。以“跨境电商多语言客服Agent”的Graph为例简化[INPUT] ↓ (map: user_input → /raw_input) [DETECT_LANGUAGE] → (output: {lang: es, confidence: 0.92}) ↓ (map: lang → /context/lang, confidence → /context/lang_confidence) [TRANSLATE_TO_EN] → (output: {en_text: I want to return item ABC123}) ↓ (map: en_text → /context/en_query) [EXTRACT_ORDER_ID] → (output: {order_id: ABC123}) ↓ (map: order_id → /context/order_id) [FETCH_ORDER_DETAILS] → (output: {status: shipped, items: [...]}) ↓ (map: status → /context/order_status, items → /context/items) [GENERATE_RESPONSE] → (input_schema requires: /context/lang, /context/en_query, /context/order_status) ↓ (output: {response_text: Su devolución está procesando..., translated_to: es}) [TRANSLATE_BACK] → (input: response_text translated_to → call translation API)这个Graph的关键特性Schema驱动每个节点声明严格的input_schema和output_schemaGraph引擎在执行前校验路径存在性。GENERATE_RESPONSE节点若发现/context/order_status不存在立即报错MISSING_CONTEXT_PATH而非等到LLM调用失败路径寻址Context是树状结构/context/lang和/context/en_query是独立路径避免context[lang]被意外覆盖Fallback链[EXTRACT_ORDER_ID]节点配置fallback_node_idRULE_BASED_EXTRACTOR当LLM提取失败时自动调用规则引擎备用方案。4.2 Graph引擎的实现轻量级但高可靠我们没有用复杂图计算框架而是基于Python字典和JSONPath实现的轻量引擎核心代码class GraphEngine: def __init__(self, graph_def: GraphDefinition): self.graph_def graph_def self.context ContextTree() # 树状Context支持路径存取 def execute(self, initial_input: dict) - dict: # Step 1: 初始化Context self.context.set(/input, initial_input) # Step 2: 按拓扑序执行Nodes for node_id in self.graph_def.topological_order(): node_def self.graph_def.nodes[node_id] # Step 2.1: 构建Node输入从Context按路径读取 node_input {} for input_path, context_path in node_def.input_mapping.items(): try: value self.context.get(context_path) node_input[input_path] value except KeyError: raise GraphError(fMissing context path: {context_path}) # Step 2.2: 执行Node可能调用Harness/Loop/外部API try: node_output self._run_node(node_def, node_input) except NodeError as e: # 触发Fallback if node_def.fallback_node_id: node_output self._run_fallback(node_def.fallback_node_id, node_input) else: raise e # Step 2.3: 写入Context按output_mapping for output_path, context_path in node_def.output_mapping.items(): if output_path in node_output: self.context.set(context_path, node_output[output_path]) return self.context.to_dict() def _run_node(self, node_def: NodeDef, input_data: dict) - dict: if node_def.type LLM: return self.harness.invoke(input_data[prompt], contextself.context.to_dict()) elif node_def.type RULE: return rule_engine.execute(node_def.rule_id, input_data) elif node_def.type API: return api_client.call(node_def.api_url, input_data)为什么不用Airflow启动开销Airflow Worker启动需3.2秒而我们的GraphEngine实例化仅17ms适合QPS1000的场景Context粒度Airflow Task间只能传序列化字典我们的ContextTree支持毫秒级路径存取context.get(/nodes/A/result/field)动态性GraphDefinition可热更新通过Redis Pub/Sub无需重启服务。当新增小语种支持时只需推送新GraphDef5秒内生效。4.3 Graph层避坑清单让拓扑结构真正“活”起来陷阱后果我们的解法效果Context路径爆炸/nodes/step1/output/data/item_list/0/name这种路径难读难维护强制使用语义化短路径/order/id,/user/lang,/response/text禁止嵌套超过3层开发者理解成本降低70%错误率下降45%节点输出Schema不一致A节点输出{items: [...]}B节点期望{products: [...]}导致运行时KeyErrorGraph引擎在node_output写入Context前执行Schema校验if not output_schema.validate(node_output): raise SchemaValidationError上线前捕获98%的Schema不匹配问题Fallback节点无限递归A fallback to B, B fallback to A在_run_fallback()中加入深度计数器max_fallback_depth2彻底杜绝递归崩溃Fallback失败时抛出FALLBACK_CHAIN_EXHAUSTED无法追踪Context来源不知道/order/status是来自API还是Mock数据ContextTree每个节点存储source属性如api:oms/v1/orders/{id}或mock:order_statuscontext.get(/order/status, with_sourceTrue)返回(value, source)审计时可精确追溯每个字段源头满足GDPR要求提示Graph不是越复杂越好。我们曾设计过一个27节点的投研Agent Graph结果发现80%的流量只走其中5个节点。后来采用“主干插件”模式主干Graph7节点处理95%常规Case插件Graph如SEC_FILING_ANALYZER按需加载。这使平均响应时间从2.1s降至0.8s。5. 三层协同当Harness、Loop、Graph在生产环境里“打架”5.1 协同故障的典型场景与根因分析三层不是独立运行它们的交互点就是故障高发区。我们记录了线上最常发生的三类协同故障场景1Harness成功Loop卡死Graph失联现象用户发起退货Harness返回{action:REFUND,amount:299}但Loop状态停在WAITING_FOR_USER_CONFIRMATIONGraph中GENERATE_RESPONSE节点从未执行根因Loop的WAIT_FOR_USER_CONFIRMATION态要求Context中存在/user/confirmation_channel字段如wechat但Harness未将其写入Context解法在Harness的_update_context()中强制注入context[user][confirmation_channel] get_channel_from_session(session_id)同时在Graph的GENERATE_RESPONSE节点input_schema中将confirmation_channel设为requiredTrue让Graph引擎在执行前就报错而非让Loop空转。场景2Graph节点失败Harness重试Loop失控现象FETCH_ORDER_DETAILS节点因第三方API超时失败Harness按配置重试3次每次失败都触发Loop的RETRY_WITH_ENHANCED_CONTEXT态结果Loop在10秒内执行了15次耗尽API配额根因Harness的重试和Loop的重试未解耦。Harness重试是技术层面网络抖动Loop重试是业务层面需补充信息解法Harness重试仅限NetworkError对APIError如HTTP 500直接失败Loop层收到APIError后不是重试而是转入ENHANCE_CONTEXT_WITH_ALTERNATIVE_DATA态调用缓存或规则引擎获取替代数据。场景3Loop状态变更Graph Context失效现象用户在WAITING_FOR_USER_CONFIRMATION态超时Loop转入RETRY_WITH_ENHANCED_CONTEXT态但Graph中GENERATE_RESPONSE节点仍读取旧的/context/en_query未使用Loop新注入的/context/enhanced_query根因Graph的Context路径是静态绑定的Loop注入的新路径未被Graph识别解法在Loop的_execute_exit_action()中当状态变更时主动通知Graph引擎刷新Context映射graph_engine.refresh_context_mapping(new_context_paths)同时Graph节点的input_mapping支持JMESPath表达式如enhanced_query || en_query实现fallback。5.2 三层协同的黄金法则契约先行避免协同故障的唯一方法是在代码之外用机器可读的契约Contract定义三层接口。我们采用YAML契约文件# harness_contract.yaml harness_refund: input_schema: type: object properties: user_input: {type: string} context: type: object properties: user_id: {type: string} session_id: {type: string} output_schema: type: object properties: action: {enum: [REFUND, EXCHANGE, CANCEL]} amount: {type: number, minimum: 0} reason: {type: string} # loop_contract.yaml loop_refund: states: WAITING_FOR_USER_CONFIRMATION: entry_conditions: - context_path: /context/order_id exists: true exit_actions: - type: send_message channel: $.context.user.confirmation_channel # JMESPath引用 template: 请确认退款¥{{ $.output.amount }} RETRY_WITH_ENHANCED_CONTEXT: entry_conditions: - event: USER_TIMEOUT exit_actions: - type: inject_context path: /context/enhanced_query value: 用户未确认升级为自动审批流程 # graph_contract.yaml graph_refund: nodes: GENERATE_RESPONSE: input_mapping: lang: /context/user/lang query: /context/enhanced_query || /context/en_query # fallback语法 order_status: /context/order/status output_mapping: response_text: /context/response/text这套契约带来的改变开发阶段Harness开发者用jsonschema.validate()校验输出Loop开发者用jmespath.search()测试条件表达式Graph开发者用pydantic校验输入部署阶段CI流水线运行contract-compatibility-check确保Harness输出Schema与Graph输入Schema兼容否则阻断发布运维阶段当Loop状态变更时监控系统自动比对loop_contract.yaml中的exit_actions验证是否所有inject_context操作都成功执行。5.3 生产环境监控三层指标必须联动监控不能只看“Agent整体成功率”必须拆解到三层层级关键指标告警阈值根因定位技巧Harnessharness_invoke_latency_p95 2sharness_parse_failure_rate 0.5%harness_cost_per_call ¥1.2latency 3s持续5分钟parse_failure 2%持续10分钟查harness_error_code分布若OUTPUT_FORMAT_INVALID突增检查模型版本若RATE_LIMIT_EXCEEDED突增查harness_model_name维度Looploop_iteration_count_p95 3loop_state_transition_rate各态流入/流出loop_human_handoff_rate 5%iteration_count 5持续10分钟handoff_rate 10%持续30分钟看loop_state热力图若WAITING_FOR_USER_CONFIRMATION流入激增但流出停滞查消息通道健康度Graphgraph_node_execution_time_p95 800msgraph_context_path_missing_rate 0.1%graph_fallback_invocation_rate 1%node_time 1.5s持续5分钟path_missing 0.5%持续10分钟按graph_node_id分组若FETCH_ORDER_DETAILS的fallback_invocation_rate达100%说明OMS系统不可用最关键的联动监控三层延迟瀑布图。当用户投诉“响应慢”我们看Harness层耗时正常1.2s→ 排除模型问题Loop层WAITING_FOR_USER_CONFIRMATION态停留2.8s → 发现消息推送延迟Graph层无异常 → 确认非编排问题。这使MTTR平均修复时间从小时级降至分钟级。6. 从Demo到生产三层架构的演进路线图6.1 阶段1Proof of Concept1周目标验证核心逻辑可行不追求性能和稳定性。Harness用openai.ChatCompletion.create()硬编码输出用正则提取Loopwhile True:time.sleep(1)轮询用户消息Graph手写if-elif-else链节点间用全局变量传值交付物一个能跑通的Notebook证明“用LLM做XX是可能的”。实操心得此阶段必须严格限制Scope。曾有团队在POC阶段就试图接入企业微信API结果卡在OAuth认证两周。我们的做法是用Mock API返回固定JSONPOC只验证LLM逻辑API集成放到Stage 2。6.2 阶段2MVP2-4周目标可内部试用具备基础可观测性。Harness封装成Class加入基础重试、超时、日志Loop实现3个核心StatePROCESS_INPUT,WAIT_FOR_CONFIRM,GENERATE_OUTPUT用Redis存StateGraph用networkx定义DAG节点间用dict传参交付物一个Web界面产品经理可输入测试Case查看每层日志。注意此阶段必须建立“契约初稿”。哪怕只是手写YAML也要明确Harness输出字段名、Loop状态名、Graph节点ID。这能避免后期重构灾难。6.3 阶段3Production Ready6-12周目标满足SLA可灰度发布。Harness接入监控Prometheus、熔断Resilience