企业官网建设流程全解析

1. 这不是概念堆砌，而是Agent落地时每天要面对的真实战场

“Harness、Loop、Graph：Agent 工程的三层架构与生产实践全解析”——这个标题乍看像学术论文，但如果你正在真实推进一个能跑通业务闭环的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的“神经反射弧”，它不定义目标，但决定目标能否达成——它管理状态流转、重试策略、人工干预点、超时熔断、结果校验。没有可控的Loop，Agent要么死循环，要么一击即溃。
• Graph是Agent的“大脑皮层”，它不执行计算，但决定计算如何组织——它编排节点依赖、传递上下文、聚合多源结果、处理分支逻辑。没有清晰的Graph，Agent就是一堆孤立函数，无法应对复杂任务。

这篇文章不讲LLM原理，不对比各家模型API，只聚焦一件事：当你手上有业务需求、有可用模型、有开发资源，如何用Harness/Loop/Graph这三层，把Agent从Demo变成可监控、可扩缩、可迭代的生产服务。适合两类人：一是正被“Agent怎么落地”困扰的工程师，二是需要评估Agent项目可行性的技术负责人。下面所有内容，都来自我们踩过的坑、压测的数据、线上监控截图和深夜改完的配置文件。

2. Harness层：不是简单封装API，而是构建Agent的“呼吸系统”

2.1 Harness的本质是协议转换器，不是HTTP客户端

很多团队第一步就错了：把Harness当成一个“调用大模型API的工具类”。结果写了一堆model.invoke(prompt)，然后发现输出格式不稳定、token超限没预警、错误类型混乱、成本无法分摊。这就像给汽车装了个能点火的按钮，却没配油路、电路、冷却系统。

Harness真正的职责，是完成四重协议转换：

1. 业务协议 → 模型协议：把订单ID、用户画像、历史对话等业务字段，按模型要求的JSON Schema或XML结构组装成Prompt；
2. 模型协议 → 业务协议：把模型返回的自由文本，用Schema约束+正则校验+LLM后处理，强制转成{ "action": "REFUND", "amount": 299.0, "reason": "damaged" }这样的确定性结构；
3. 成本协议 → 财务协议：把input_tokens: 1247, output_tokens: 89实时换算成人民币金额（考虑不同模型、不同region的单价），并打标到业务维度（如“客服退换货场景”）；
4. 错误协议 → 运维协议：把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_ms=3000, max_retries=2等 self.parser = JSONSchemaParser(schema=REFUND_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( prompt=prompt, temperature=self.config.temperature, max_tokens=self.config.max_output_tokens, timeout=self.config.timeout_ms ) except ModelTimeoutError as e: raise HarnessError("MODEL_TIMEOUT", cause=e, context=context) except ModelRateLimitError as e: raise HarnessError("RATE_LIMIT_EXCEEDED", cause=e, context=context) # 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", cause=e, context=context) else: # 如金额超范围，尝试LLM后处理修复 repaired = self._repair_with_llm(raw_response, REFUND_SCHEMA) if repaired: parsed = repaired else: raise HarnessError("OUTPUT_SEMANTIC_INVALID", cause=e, context=context) # Step 4: 成本核算与审计 cost = self.cost_tracker.calculate( input_tokens=len(prompt), output_tokens=len(raw_response), model_name=self.model_client.model_name ) audit_log = AuditLog( trace_id=context.get('trace_id'), harness_version="v2.3.1", cost=cost, input_hash=hashlib.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层避坑指南：那些文档里不会写的细节

注意：不要在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(definition=LOOP_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(failed=True, error_code=e.code, context=context) # Step 2: 更新Context（关键！Loop的生命线） context = self._update_context(context, action_result) # Step 3: 根据Action结果和Guard条件，决定下一个State next_state = self.state_machine.transition( current_state=state, event=action_result.event, # 如"USER_CONFIRMED", "API_TIMEOUT" context=context, 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(success=True, final_context=context) else: return LoopResult(failed=True, error_code="LOOP_MAX_ITERATION_EXCEEDED", context=context) 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()返回event="USER_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.6（Harness层需返回置信度）。

第二，放弃必须有“移交路径”。放弃不是结束，而是转交。我们的标准移交路径：

1. 生成结构化摘要（含时间线、已尝试步骤、失败原因）；
2. 推送至指定队列（如human_handoff_queue）；
3. 向用户发送移交确认消息（“您的问题已转交高级专员，将在2小时内回复”）；
4. 记录移交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_time=302s > 300s", "retry_count=0"] } 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_id；
• Edge（边）：不是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_schema，Graph引擎在执行前校验路径存在性。GENERATE_RESPONSE节点若发现/context/order_status不存在，立即报错MISSING_CONTEXT_PATH，而非等到LLM调用失败；
• 路径寻址：Context是树状结构，/context/lang和/context/en_query是独立路径，避免context['lang']被意外覆盖；
• Fallback链：[EXTRACT_ORDER_ID]节点配置fallback_node_id="RULE_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(f"Missing 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"], context=self.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，适合QPS>1000的场景；
• Context粒度：Airflow Task间只能传序列化字典，我们的ContextTree支持毫秒级路径存取（context.get("/nodes/A/result/field")）；
• 动态性：GraphDefinition可热更新（通过Redis Pub/Sub），无需重启服务。当新增小语种支持时，只需推送新GraphDef，5秒内生效。

4.3 Graph层避坑清单：让拓扑结构真正“活”起来

提示：Graph不是越复杂越好。我们曾设计过一个27节点的投研Agent Graph，结果发现80%的流量只走其中5个节点。后来采用“主干+插件”模式：主干Graph（7节点）处理95%常规Case，插件Graph（如SEC_FILING_ANALYZER）按需加载。这使平均响应时间从2.1s降至0.8s。

5. 三层协同：当Harness、Loop、Graph在生产环境里“打架”

5.1 协同故障的典型场景与根因分析

三层不是独立运行，它们的交互点就是故障高发区。我们记录了线上最常发生的三类协同故障：

场景1：Harness成功，Loop卡死，Graph失联

• 现象：用户发起退货，Harness返回{"action":"REFUND","amount":299}，但Loop状态停在WAITING_FOR_USER_CONFIRMATION，Graph中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设为required=True，让Graph引擎在执行前就报错，而非让Loop空转。

场景2：Graph节点失败，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态，调用缓存或规则引擎获取替代数据。

场景3：Loop状态变更，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整体成功率”，必须拆解到三层：

最关键的联动监控：三层延迟瀑布图。当用户投诉“响应慢”，我们看：

• Harness层耗时正常（<1.2s）→ 排除模型问题；
• Loop层WAITING_FOR_USER_CONFIRMATION态停留2.8s → 发现消息推送延迟；
• Graph层无异常 → 确认非编排问题。
  这使MTTR（平均修复时间）从小时级降至分钟级。

6. 从Demo到生产：三层架构的演进路线图

6.1 阶段1：Proof of Concept（1周）

目标：验证核心逻辑可行，不追求性能和稳定性。

• Harness：用openai.ChatCompletion.create()硬编码，输出用正则提取；
• Loop：while True:+time.sleep(1)轮询用户消息；
• Graph：手写if-elif-else链，节点间用全局变量传值；
• 交付物：一个能跑通的Notebook，证明“用LLM做XX是可能的”。

实操心得：此阶段必须严格限制Scope。曾有团队在POC阶段就试图接入企业微信API，结果卡在OAuth认证两周。我们的做法是：用Mock API返回固定JSON，POC只验证LLM逻辑，API集成放到Stage 2。

6.2 阶段2：MVP（2-4周）

目标：可内部试用，具备基础可观测性。

• Harness：封装成Class，加入基础重试、超时、日志；
• Loop：实现3个核心State（PROCESS_INPUT,WAIT_FOR_CONFIRM,GENERATE_OUTPUT），用Redis存State；
• Graph：用networkx定义DAG，节点间用dict传参；
• 交付物：一个Web界面，产品经理可输入测试Case，查看每层日志。

注意：此阶段必须建立“契约初稿”。哪怕只是手写YAML，也要明确Harness输出字段名、Loop状态名、Graph节点ID。这能避免后期重构灾难。

6.3 阶段3：Production Ready（6-12周）

目标：满足SLA，可灰度发布。

• Harness：接入监控（Prometheus）、熔断（Resilience