Agent生产落地三阶架构：Harness、Loop、Graph工程实践

1. 这不是又一个“架构图”，而是Agent落地时每天要面对的真实战场

你有没有过这样的经历：花两周搭好一个Agent流程，本地跑通、demo惊艳，结果一上生产环境就卡在第三步——不是LLM响应慢，不是工具调用失败，而是整个执行链路像被无形的手掐住喉咙：任务分发不均、状态丢失、重试逻辑失控、监控日志里全是“unknown error”；更糟的是，当业务方突然要求加个“用户中断后自动保存上下文并续办”的功能，你翻遍代码发现——根本没地方插。这不是技术能力问题，是架构层缺失导致的系统性失能。

标题里的Harness、Loop、Graph，不是三个并列概念，而是一条从“单点控制”走向“系统韧性”的演进路径，是我在过去三年带团队交付17个Agent生产项目（覆盖金融风控、政务问答、工业设备运维三大场景）中，踩坑、复盘、重构、再验证后沉淀下来的三层骨架。它不讲虚的“智能体范式”，只解决三类硬问题：

• Harness层：怎么让一个LLM调用、一次API请求、一段Python脚本，变成可编排、可监控、可熔断的“原子服务”？
• Loop层：当Agent需要多轮交互、状态流转、条件分支、人工介入时，如何避免写一堆if-else和全局变量，让逻辑既清晰又抗压？
• Graph层：当业务复杂到几十个Agent协同、数据在多个系统间流转、故障需要跨服务溯源时，怎么让整个系统“看得见、控得住、救得快”？

这三个词在热搜里常被割裂讨论：“harness和agent区别”、“loop engineering介绍”、“snap graph builder”……但真实生产中，它们是咬合传动的齿轮。Harness失效，Loop就是空中楼阁；Loop设计僵化，Graph再漂亮也是静态沙盘。本文不堆砌论文术语，不复述LLM原理，只讲我们怎么用这三层，在银行信贷审批Agent里把平均响应时间从8.2秒压到1.4秒，在政务热线Agent中实现99.99%的会话状态零丢失，在工业设备Agent中将故障定位耗时从小时级缩短到分钟级。所有方案都已在K8s集群+Python3.11+LangChain v0.1.16+Redis 7.2环境下稳定运行超400天，配置细节、参数取舍、避坑清单全部公开。

2. Harness层：把“调用LLM”变成“调度原子服务”的工程化封装

2.1 为什么不能直接调openai.ChatCompletion.create()？

新手常犯的第一个错误，就是把LLM调用当成普通函数调用。我见过最典型的反模式：在一个Flask路由里，直接response = client.chat.completions.create(...)，然后把response.choices[0].message.content塞进返回体。表面看没问题，但上线后立刻暴露三重脆弱性：

• 无熔断：OpenAI API偶尔503，整个HTTP请求直接超时，前端白屏；
• 无追踪：用户投诉“回答不对”，你查日志只能看到“request_id: abc123”，但不知道这次调用用了哪个prompt模板、temperature设为多少、是否触发了tool call；
• 无复用：同一个system prompt在10个不同接口里硬编码，改一处漏九处。

Harness的本质，就是给每一次“非确定性计算”（LLM推理、外部API、长耗时脚本）套上确定性的工程外壳。它不是框架，而是一组约定：输入标准化、输出结构化、过程可观测、失败可兜底。

2.2 Harness的核心契约：四要素不可缺

一个合格的Harness必须同时满足以下四个契约，缺一不可。我们在金融风控Agent中定义的Harness标准如下（已抽象为Pydantic模型）：

from pydantic import BaseModel, Field from typing import Optional, Dict, Any class HarnessInput(BaseModel): # 必须显式声明输入来源，禁止隐式依赖全局变量 context: Dict[str, Any] = Field(default_factory=dict) # 用户会话、历史消息、业务实体等 config: Dict[str, Any] = Field(default_factory=dict) # temperature, max_tokens, tools等LLM参数 metadata: Dict[str, str] = Field(default_factory=dict) # trace_id, user_id, request_source等追踪字段 class HarnessOutput(BaseModel): # 输出必须结构化，禁止返回原始字符串 result: str = "" # 主要文本结果 tool_calls: list[dict] = [] # 工具调用指令（如{"name": "get_account_balance", "args": {"account_id": "123"}}） metadata: Dict[str, Any] = Field(default_factory=dict) # cost, latency, model_used等性能指标 class HarnessError(BaseModel): code: str = "HARNESSE_ERROR" # 错误码，用于分类告警 message: str = "" # 用户友好的错误提示 detail: str = "" # 技术细节，仅日志记录 retryable: bool = True # 是否允许自动重试 class Harness(BaseModel): name: str # 唯一标识，如 "credit_score_evaluator" input_schema: HarnessInput output_schema: HarnessOutput error_schema: HarnessError version: str = "1.0.0" # 语义化版本，支持灰度发布

提示：这个契约强制开发者思考“这个LLM调用到底在做什么业务动作”。比如name="credit_score_evaluator"比name="llm_call_3"有意义得多——它直接关联到业务域，方便后续在Graph层做拓扑分析。

2.3 实战：构建一个带熔断与缓存的Harness实例

以政务热线Agent中的“政策条款解析”功能为例，其Harness需满足：

• 对高频查询（如“低保申请条件”）缓存结果，降低LLM调用频次；
• 当OpenAI连续3次超时，自动降级到本地微调模型（Llama3-8B）；
• 每次调用记录token消耗，超预算自动截断。

我们基于tenacity和redis-py实现核心逻辑（已脱敏）：

# policy_parser_harness.py import redis from tenacity import retry, stop_after_attempt, wait_exponential, retry_if_exception_type from openai import AsyncOpenAI from .schemas import HarnessInput, HarnessOutput, HarnessError class PolicyParserHarness: def __init__(self, redis_client: redis.Redis, openai_client: AsyncOpenAI): self.redis = redis_client self.openai = openai_client self.fallback_model = "meta-llama/Meta-Llama-3-8B-Instruct" # 本地部署的Ollama模型 @retry( stop=stop_after_attempt(3), wait=wait_exponential(multiplier=1, min=1, max=10), retry=retry_if_exception_type((TimeoutError, ConnectionError)) ) async def execute(self, input_data: HarnessInput) -> HarnessOutput | HarnessError: # Step 1: 尝试从Redis获取缓存（key = md5(policy_text + config)） cache_key = self._gen_cache_key(input_data.context.get("policy_text", ""), input_data.config) cached = self.redis.get(cache_key) if cached: return HarnessOutput.model_validate_json(cached) # Step 2: 构建Prompt（严格遵循schema，禁止拼接字符串） system_prompt = "你是一名政务政策解读专家，请严格按JSON格式输出..." user_content = f"政策原文：{input_data.context['policy_text']}\n要求：{input_data.context.get('requirement', '')}" try: response = await self.openai.chat.completions.create( model="gpt-4o", messages=[{"role": "system", "content": system_prompt}, {"role": "user", "content": user_content}], temperature=input_data.config.get("temperature", 0.3), max_tokens=input_data.config.get("max_tokens", 512), response_format={"type": "json_object"} # 强制JSON输出，便于解析 ) # Step 3: 结构化解析，提取result和tool_calls raw_content = response.choices[0].message.content parsed = json.loads(raw_content) output = HarnessOutput( result=parsed.get("summary", ""), tool_calls=parsed.get("tool_calls", []), metadata={ "cost": self._calc_cost(response.usage), "latency_ms": response.usage.total_tokens * 0.8, # 简化估算 "model_used": "gpt-4o" } ) # Step 4: 写入缓存（TTL=3600秒，高频政策不变） self.redis.setex(cache_key, 3600, output.model_dump_json()) return output except Exception as e: # Step 5: 降级处理（此处简化，实际需调用本地Ollama API） if "timeout" in str(e).lower(): return await self._fallback_execute(input_data) else: return HarnessError( code="OPENAI_PARSE_ERROR", message="政策解析失败，请稍后重试", detail=str(e), retryable=False )

注意：这里response_format={"type": "json_object"}是关键。它让GPT-4o原生输出JSON，避免了正则匹配或json.loads()失败的风险。我们在测试中发现，未启用此参数时，JSON解析失败率高达12%，启用后降至0.3%。

2.4 Harness层的避坑清单：那些文档里不会写的细节

• 缓存Key设计陷阱：不要用str(input_data)作为cache key！Python字典顺序不确定，{"a":1,"b":2}和{"b":2,"a":1}序列化后字符串不同。必须用json.dumps(input_data.dict(), sort_keys=True)。
• 熔断阈值怎么定？我们实测：对OpenAI API，连续2次超时（>15s）即触发熔断，而非默认的3次。因为第3次失败时，用户已刷新页面，重试无意义。
• 降级模型的选择逻辑：本地模型不能简单“备用”。我们在政务场景中，对政策类问题用Llama3-8B，对地址解析类问题用专门微调的TinyBERT，因为领域适配度比模型大小更重要。
• Metadata字段必须预留扩展位：output.metadata里我们强制保留"trace_id"和"span_id"字段，即使当前不用，为后续接入Jaeger埋点留出空间。曾有项目因没预留，后期全量重构。

3. Loop层：告别状态管理地狱，用有限状态机驱动Agent决策流

3.1 为什么“while True:”是Agent生产的最大定时炸弹？

很多教程教用while True:循环实现多轮对话，看似简单：

# 危险示范！ while True: user_input = get_user_input() llm_response = call_llm(user_input, history) if llm_response.contains("final_answer"): break history.append(llm_response)

问题在于：这个循环完全脱离系统管控。当用户在第5轮说“算了，先帮我查下账户余额”，程序无法优雅中断当前流程；当LLM在第3轮意外返回空字符串，整个循环卡死；当并发1000个会话，每个都持有一个history列表，内存暴涨。

Loop层的核心使命，是把Agent的“思考过程”变成可调度、可暂停、可回溯的状态机。它不关心LLM怎么想，只定义“想的步骤”和“每步的守门人”。

3.2 Loop的三种形态：State Machine、Workflow、Reactive Stream

根据业务复杂度，我们采用三种Loop实现，而非一种“万能框架”：

实操心得：90%的政务和金融Agent，用FSM足够。我们在铜陵市12345热线项目中，用transitions定义了7个状态（idle,intent_recognized,policy_searching,answer_generating,answer_reviewing,answered,escalated），状态图打印出来只有A4纸大小，但覆盖了全部237个业务场景。

3.3 FSM Loop实战：政务热线Agent的状态机设计

以“社保缴费查询”子流程为例，其FSM定义如下（精简版）：

from transitions import Machine class SocialSecurityLoop: states = ['idle', 'user_query_received', 'identity_verified', 'query_executed', 'answer_generated', 'answered', 'error'] transitions = [ # 初始接收用户query {'trigger': 'receive_query', 'source': 'idle', 'dest': 'user_query_received', 'conditions': 'is_valid_query'}, # 身份核验（调用Harness） {'trigger': 'verify_identity', 'source': 'user_query_received', 'dest': 'identity_verified', 'conditions': 'call_id_verify_harness', 'after': 'log_verification_result'}, # 执行查询（调用Harness） {'trigger': 'execute_query', 'source': 'identity_verified', 'dest': 'query_executed', 'conditions': 'call_query_harness', 'after': 'cache_query_result'}, # 生成回答（调用Harness） {'trigger': 'generate_answer', 'source': 'query_executed', 'dest': 'answer_generated', 'conditions': 'call_answer_gen_harness'}, # 返回结果 {'trigger': 'send_answer', 'source': 'answer_generated', 'dest': 'answered', 'after': 'record_metrics'}, # 任意状态可跳转error {'trigger': 'handle_error', 'source': '*', 'dest': 'error', 'before': 'log_error'}, ] def __init__(self): self.machine = Machine(model=self, states=SocialSecurityLoop.states, transitions=SocialSecurityLoop.transitions, initial='idle') self.current_session = {} def is_valid_query(self): return "社保" in self.current_session.get("user_input", "") and len(self.current_session.get("user_input", "")) > 5 def call_id_verify_harness(self): # 调用id_verify_harness，成功则返回True harness_input = HarnessInput( context={"id_number": self.current_session.get("id_number")}, config={"temperature": 0.0} ) result = asyncio.run(id_verify_harness.execute(harness_input)) self.current_session["id_verified"] = result.result == "verified" return result.result == "verified" # ... 其他条件方法同理

关键设计点：

• 状态迁移必须有明确触发器（trigger）和守门条件（conditions），避免隐式跳转；
• 所有业务逻辑放在conditions和after钩子里，state本身只存状态，不存数据；
• self.current_session是唯一的数据载体，所有Harness调用都从这里读写，彻底解耦状态与业务。

3.4 Loop层的并发与状态持久化：别让Redis成为瓶颈

当Loop运行在高并发场景（如银行日均50万次信贷评估），状态存储是生死线。我们踩过的坑：

• 错误方案：用Redis Hash存每个session的完整状态对象。问题：每次状态变更都要HGETALL+HSET，网络IO成瓶颈，QPS卡在1200。
• 正确方案：用Redis Stream + Lua脚本实现原子状态更新。每个session对应一个Stream，状态变更作为一条消息写入，消费端（Loop Engine）监听Stream并更新内存状态。实测QPS提升至8500+。

核心Lua脚本（update_state.lua）：

-- KEYS[1] = stream_key, ARGV[1] = session_id, ARGV[2] = new_state, ARGV[3] = metadata_json local stream_key = KEYS[1] local session_id = ARGV[1] local new_state = ARGV[2] local metadata = cjson.decode(ARGV[3]) -- 构建消息体 local msg = { session_id = session_id, state = new_state, timestamp = tonumber(redis.call('TIME')[1]), metadata = metadata } -- 写入Stream（XADD） redis.call('XADD', stream_key, '*', 'data', cjson.encode(msg)) -- 更新最新状态到Hash（供快速查询） redis.call('HSET', 'session_state:' .. session_id, 'state', new_state, 'updated_at', msg.timestamp) return 1

实操心得：Stream的XREADGROUP消费必须设置COUNT 100批量读取，否则单条消息处理的网络开销会吃掉30%性能。我们在压测中发现，COUNT 10和COUNT 100的吞吐量相差2.3倍。

4. Graph层：当Agent不再是孤岛，而是可编排、可观测、可治理的网络节点

4.1 Graph不是画出来的，是跑出来的：从静态拓扑到动态血缘

很多团队花一周时间用draw.io画出“Agent架构图”，包含LLM、Tool、Database、Cache等节点，箭头标着数据流向。这图在项目启动时有用，但上线三天后就失效——因为没人更新它。真正的Graph层，必须是由生产流量自动生成、实时更新、可交互查询的动态血缘图。

我们在工业设备Agent中，Graph层承担三重角色：

• 编排中枢：根据设备类型（PLC/DCS/SCADA）自动选择不同的Agent组合；
• 故障雷达：当某台设备报警，Graph能1秒内定位到影响的3个Agent、2个数据库表、1个缓存Key；
• 容量沙盘：模拟“增加1000台新设备”对各Agent的负载影响，提前扩容。

4.2 Graph的两种构建范式：声明式 vs. 运行时注入

我们采用混合方案：基础设施用声明式，业务逻辑用运行时注入。

4.3 运行时Graph构建：基于OpenTelemetry的轻量级实现

放弃Jaeger/SkyWalking等重型APM，自研轻量Graph Collector，核心逻辑：

• Agent侧：每个Harness执行前后，向本地UDP端口发送Span（无需修改业务代码，通过装饰器注入）；
• Collector侧：用Go编写，接收Span、去重、聚合、构建邻接表，每5秒更新一次Graph快照；
• Query侧：提供GraphQL API，支持findPath(from: "credit_evaluator", to: "risk_db")等查询。

关键Span结构（简化）：

{ "trace_id": "0x1a2b3c...", "span_id": "0x4d5e6f...", "parent_span_id": "0x7g8h9i...", // 形成调用链 "service_name": "credit_agent", "operation_name": "evaluate_credit_score", "start_time": 1712345678.123, "end_time": 1712345679.456, "attributes": { "harness.name": "credit_score_evaluator", "harness.input_hash": "md5(...)", "harness.output_size": 1245, "db.query": "SELECT * FROM credit_risk WHERE id=?" } }

提示：attributes字段是Graph价值所在。我们强制要求所有Harness在output.metadata中注入"harness.name"，并在Span中透传。这样Graph就能知道“这个DB查询是由哪个Harness发起的”，实现真正端到端血缘。

4.4 Graph层的实战价值：故障定位从“查日志”到“点图形”

以某次生产事故为例：

• 现象：政务热线Agent的“公积金提取”功能，响应时间从1.2秒突增至15秒；
• 传统排查：登录12台服务器，grep日志，逐个检查OpenAI调用、Redis连接、MySQL慢查询——耗时47分钟；
• Graph排查：在Graph UI中输入findPath(from: "housing_fund_extractor", to: "policy_db")，立即显示路径：
  housing_fund_extractor → policy_parser_harness → policy_db
  点击policy_db节点，显示其最近10分钟TP99从80ms飙升至2100ms；
  再点击该节点的incoming_edges，发现92%流量来自policy_parser_harness；
  最终定位：policy_parser_harness的缓存Key生成逻辑有bug，导致缓存击穿。

整个过程耗时83秒。Graph的价值，不在于画得多美，而在于让故障从“大海捞针”变成“按图索骥”。

5. 生产实践全解析：三层架构如何协同作战

5.1 场景还原：银行信贷审批Agent的三层联动

我们以一个真实案例，展示Harness、Loop、Graph如何咬合工作：

业务需求：用户上传身份证照片，Agent需完成：①OCR识别身份证号；②调用公安接口核验身份；③查询央行征信报告；④综合评分并返回结果。

Harness层就绪：

• ocr_harness：封装腾讯云OCR SDK，输入图片base64，输出结构化身份证信息；
• id_verify_harness：封装公安核验API，输入身份证号，输出核验结果；
• credit_report_harness：封装央行征信接口，输入身份证号，输出加密报告；
• score_calculator_harness：本地Python脚本，输入OCR+核验+征信数据，输出评分。

Loop层编排：

定义FSM状态：idle→ocr_received→id_verified→report_fetched→scored→result_sent。
每个状态迁移由对应Harness的成功执行触发。例如，ocr_received状态的next_state条件是：ocr_harness.execute(...).result.id_number is not None。

Graph层赋能：

• 编排：当用户上传图片，Graph根据图片尺寸（<1MB走OCR，>1MB走人工审核通道）动态选择Loop路径；
• 监控：Graph实时统计各Harness的P95延迟，当id_verify_harness延迟>3s，自动告警并降级到备用核验渠道；
• 治理：每月生成Graph报告，显示credit_report_harness调用量占总流量62%，但成本占89%，推动团队与央行协商接口优化。

实测效果：上线后，单次审批平均耗时从12.7秒降至2.3秒，人工干预率从18%降至3.2%，系统可用性达99.995%。

5.2 部署架构：三层如何物理落地

我们的生产部署采用“分层隔离、网关统管”原则：

┌─────────────────────────────────────────────────────────────────────┐ │ Client (Web/App) │ └─────────────────────────────────────────────────────────────────────┘ ↓ HTTPS ┌─────────────────────────────────────────────────────────────────────┐ │ API Gateway (Envoy) │ │ • 统一路由、鉴权、限流 │ │ • 注入trace_id到所有下游请求 │ └─────────────────────────────────────────────────────────────────────┘ ↓ gRPC ┌─────────────────────────────────────────────────────────────────────┐ │ Agent Orchestrator (Go) │ │ • 接收Gateway请求，解析业务意图 │ │ • 根据Graph查询选择最优Loop路径 │ │ • 启动对应Loop实例（每个Loop独占goroutine） │ └─────────────────────────────────────────────────────────────────────┘ ↓ gRPC ┌─────────────────────────────────────────────────────────────────────┐ │ Harness Executors (Python, K8s Deployment) │ │ • 每个Harness独立Pod，资源隔离 │ │ • 内置Redis连接池、OpenAI客户端、本地模型缓存 │ │ • 上报Span到Graph Collector │ └─────────────────────────────────────────────────────────────────────┘ ↓ ┌─────────────────────────────────────────────────────────────────────┐ │ External Systems (DB, Cache, API) │ └─────────────────────────────────────────────────────────────────────┘

关键设计：

• Orchestrator不碰业务逻辑，只做路由和状态机调度，保证高可用（Go编写，CPU占用<5%）；
• Harness Executor是无状态的，所有状态存在Loop层，Executor重启不影响会话；
• Graph Collector独立部署，不与业务Pod共用资源，避免影响主流程。

5.3 性能压测与容量规划：三层各自的瓶颈在哪？

我们对三层分别进行压测，结论颠覆常识：

实操心得：Loop层才是真正的性能咽喉。很多人以为LLM调用最慢，其实Harness只是转发，真正耗时的是状态判断、数据转换、条件分支。我们在压测中发现，Loop层CPU使用率在QPS>5000时陡增，而Harness层CPU始终<30%。因此，优化重点永远是Loop——精简状态、减少I/O、预计算。

6. 常见问题与排查技巧实录：那些深夜救火时的真实记录

6.1 Harness层典型问题速查

6.2 Loop层致命陷阱与修复

• 陷阱1：状态机“假死”
  现象：Loop卡在user_query_received状态，不触发verify_identity。
  原因：is_valid_query()条件中用了"社保" in user_input，但用户输入是“社保存款”，中文分词未生效。
  修复：条件函数中加入jieba分词，"社保" in jieba.lcut(user_input)。

• 陷阱2：并发下的状态错乱
  现象：两个用户A/B同时查询，A的结果出现在B的响应里。
  原因：self.current_session是类属性，非实例属性。
  修复：在__init__中改为self.current_session = {}，确保每个Loop实例独享状态。

• 陷阱3：无限重试循环
  现象：Loop在error状态反复触发handle_error，日志刷屏。
  原因：handle_error的after钩子中调用了另一个Harness，该Harness又失败。
  修复：在handle_error中增加计数器，if self.error_count > 3: self.to_idle()。

6.3 Graph层隐蔽Bug排查

• 问题：Graph显示某Harness调用次数为0，但日志证明它每天调用2万次
  原因：该Harness的Span中service_name写成了"credit_agent_v2"，而Graph Collector只采集"credit_agent"前缀。
  修复：统一规范service_name命名规则，用CI检查所有Harness代码。

• 问题：findPath查询超时
  原因：Graph邻接表未建立索引，10万节点时查询复杂度O(N²)。
  修复：用Redis Graph模块替代自研邻接表，GRAPH.QUERY g "MATCH (a)-[r]->(b) WHERE a.name='x' RETURN b"毫秒级响应。

• 问题：血缘图中出现“幽灵节点”
  原因：某Harness异常退出，未发送ENDSpan，Collector将其视为长活Span持续计入。
  修复：Collector增加心跳检测，30秒无新Span则自动标记为stale。

6.4 三层协同故障：一个经典案例

现象：用户反馈“提交贷款申请后，页面一直转圈，3分钟后显示‘系统繁忙’”。
排查路径：

1. Graph层：findPath(from: "loan_submitter", to: "credit_evaluator")显示路径正常；
2. Loop层：查loan_submitter状态机日志，发现卡在credit_evaluator调用后，未进入scored状态；
3. Harness层：查credit_evaluator的Span，发现end_time缺失，duration为0；
4. 深入：抓包发现credit_evaluatorPod的/healthz返回503，但K8s liveness probe未触发重启；
5. 根因：Harness Executor的健康检查逻辑有bug，if time.time() - last_success < 300: return 200，但last_success未初始化，恒为0。

教训：三层必须有统一的健康信号标准。我们现在强制所有Harness Executor暴露/healthz，返回{"status":"ok","harnesses":["ocr","id_verify"]}，Orchestrator据此决定是否路由流量。

7. 我在实际项目中总结的三条铁律

第一，Harness不是越薄越好，而是越“契约”越好。曾有个团队追求极致轻量，Harness只做参数透传，结果三个月后，12个Harness的config字段名五花八门（temp,temperature,temp_val），Loop层不得不写一堆映射逻辑。后来我们强制推行HarnessInputSchema，新增字段必须提PR，由架构委员会评审——看似慢，实则省下87%的联调时间。

第二，Loop的状态数不是越少越好，而是越“业务语义”越好。有团队把所有状态压缩成running/done/failed三个，结果运维无法区分“征信查询失败”和“OCR识别失败”，告警一律发给同一组人。现在我们要求每个状态名必须是动宾短语（identity_verified,report_fetched），且每个状态必须有对应的监控指标（loop_state_duration_seconds{state="identity_verified"}）。

第三，Graph的价值不在“画出来”，而在“动起来”。最初我们花两周开发Graph UI，结果业务方只看一眼就扔在一边。直到我们把Graph能力嵌入到CI/CD流水线：每次代码提交，自动检测