从零构建产品级AI Agent Harness：工程实践与核心架构解析

1. 项目概述：什么是产品级 Agent Harness？

如果你关注过近一两年的AI应用开发，尤其是围绕大语言模型（LLM）构建的智能体（Agent），那么“Harness”这个词出现的频率一定不低。它不像“框架”或“平台”那样宏大，也不像“工具包”那样零散。你可以把它理解为一个**“缰绳”或“约束装置”**——它的核心目标不是提供无限的可能性，而是将强大但不可控的AI能力，安全、可靠、可预测地“套”到具体的产品工作流中。

在前两篇中，我们探讨了Agent的基础概念和核心组件，比如工具调用（Tool Calling）、规划（Planning）与记忆（Memory）。但当你真正要把这些组件组装成一个能上线、能服务真实用户、能扛住生产环境压力的“产品”时，你会发现理论和demo之间存在巨大的鸿沟。这就是“产品级Agent Harness”要解决的问题：它是一套工程实践、设计模式和基础设施的集合，确保你的Agent不是实验室里的玩具，而是商业环境中的可靠员工。

简单来说，产品级Harness关注的是：稳定性、可观测性、成本控制、用户体验和迭代效率。它回答的是“如何让这个聪明的AI助手不胡说八道、不突然宕机、不烧光预算，并且能越用越好”的问题。这个系列第三篇，我们将深入核心，从零开始，动手搭建一个具备产品级潜力的Agent Harness原型，聚焦于最关键的执行与评估循环。

2. 核心架构设计：从链式思维到循环思维

在构建产品级Harness时，首要任务是摒弃简单的“输入-输出”链式思维。一个初级Agent的实现可能像这样：用户提问 -> LLM思考 -> 调用工具 -> 返回结果。这条链非常脆弱，任何环节出错（比如工具调用失败、LLM输出格式错误）都会导致整个流程崩溃，给用户一个糟糕的体验。

产品级Harness需要引入循环思维和韧性设计。其核心架构通常包含以下几个层次：

2.1 控制层：Orchestrator（编排器）

这是Harness的大脑。它不直接处理LLM调用或工具执行，而是负责任务的分解、流程的调度和异常的处理。一个典型的Orchestrator需要决定：

• 任务类型判断：用户请求是简单查询，还是需要多步执行的复杂任务？
• 规划生成与调整：根据当前状态和记忆，生成或调整下一步的执行计划。
• 执行决策：在当前步骤，是调用工具A，还是需要先向用户澄清问题？
• 循环控制：判断当前结果是否满足要求，是否需要重试、回退或转入人工流程。

在实现上，Orchestrator本身可以是一个轻量级的LLM调用（使用小模型以控制成本），也可以是一套基于规则的决策树。我们的原型将采用后者，以强调确定性和可调试性。

2.2 执行层：Tool Executor（工具执行器）

这是Harness的双手。它负责安全、隔离地执行具体的工具（函数）。产品级要求意味着：

• 沙箱环境：工具执行必须在受控的沙箱中，防止对主系统造成破坏（如执行任意代码、删除文件）。
• 超时与资源限制：每个工具调用必须有严格的超时时间和资源（CPU/内存）上限。
• 输入验证与清理：在执行前，对LLM生成的工具参数进行严格的类型和范围校验，防止注入攻击。
• 标准化输出：无论工具内部如何实现，对外输出必须统一为结构化的格式（如JSON），包含success、result、error_message等字段。

2.3 状态与记忆层：State Manager（状态管理器）

Agent是有状态的。它需要记住对话历史、已执行的操作和中间结果。产品级Harness的状态管理不能简单地将整个对话历史每次都塞给LLM（有上下文长度限制且成本高），而需要智能的摘要和检索。

• 短期记忆：保存当前会话的完整上下文，用于连贯性。
• 长期记忆：将历史会话的关键信息（如用户偏好、决策逻辑、执行结果）向量化后存入数据库，支持在后续会话中快速检索关联。
• 执行状态：保存多步任务当前的进度、已产生的中间数据，确保在中断（如网络超时）后能够恢复。

2.4 评估与安全层：Guardrails（护栏）

这是产品级的“安全带”和“质检员”。它在Agent输出最终结果前和最终结果后进行拦截和检查。

• 输入过滤：检查用户输入是否包含恶意提示、敏感信息或超出服务范围的内容。
• 过程监控：在每一步执行后，评估工具调用的结果是否合理、是否偏离目标。例如，一个查询天气的Agent突然尝试调用“发送邮件”工具，这应该被立即阻止。
• 输出校验：对LLM生成的最终答案进行事实性核查、毒性检测、格式合规性检查等。例如，确保生成的代码没有安全漏洞，确保提供的建议符合伦理规范。

我们的原型将重点实现一个包含Orchestrator、Tool Executor和基础Guardrails的简化循环系统。

3. 实战构建：一个任务执行Harness原型

让我们以一个具体的场景来构建原型：“智能数据查询助手”。用户可以用自然语言描述复杂的数据查询需求，Agent需要理解需求，将其转化为一系列数据库查询工具调用，并整合结果返回。

3.1 定义工具集与状态Schema

首先，明确Agent能做什么。我们定义三个核心工具：

1. query_database(sql_query: str) -> List[Dict]: 执行SQL查询。
2. get_table_schema(table_name: str) -> Dict: 获取指定数据表的字段结构。
3. explain_query_result(data: List[Dict]) -> str: 用自然语言解释查询结果。

接下来，定义整个系统的执行状态Schema，这将是贯穿循环的核心数据结构：

from pydantic import BaseModel, Field from typing import Dict, Any, List, Optional class AgentState(BaseModel): """Agent执行状态""" user_input: str # 原始用户输入 parsed_intent: Optional[str] = None # 解析后的用户意图 current_plan: List[str] = [] # 当前执行计划，如 [“get_schema”, “query_db”] completed_steps: List[Dict] = [] # 已完成的步骤及其结果 available_tools: List[str] = Field(default_factory=lambda: ["query_database", "get_table_schema", "explain_query_result"]) max_iterations: int = 10 # 最大循环次数，防止死循环 iteration_count: int = 0 # 当前迭代次数 final_answer: Optional[str] = None # 最终给用户的答案 error: Optional[str] = None # 执行过程中的错误信息

使用Pydantic进行数据验证能极大提高系统的健壮性。

3.2 实现编排器（Orchestrator）

我们的编排器基于规则，它根据当前状态决定下一步动作。这是一个简化的决策逻辑：

class RuleBasedOrchestrator: def decide_next_action(self, state: AgentState) -> str: """ 根据当前状态决定下一步动作。 返回动作类型：'need_clarification', 'execute_tool', 'generate_final_answer', 'error' """ state.iteration_count += 1 if state.iteration_count > state.max_iterations: return 'error' # 超过最大迭代次数 if not state.parsed_intent: # 第一步：解析用户意图 return 'parse_intent' elif not state.current_plan: # 第二步：生成执行计划 return 'generate_plan' elif state.final_answer is not None: # 已有最终答案，结束 return 'finished' elif state.error: # 发生错误，结束 return 'error' else: # 执行计划中的下一步 # 这里简化逻辑：如果已完成步骤数小于计划长度，则执行工具 if len(state.completed_steps) < len(state.current_plan): return 'execute_tool' else: # 计划已完成，生成最终答案 return 'generate_final_answer'

这个编排器非常基础，但关键在于它建立了清晰的状态转移逻辑。在实际产品中，这里的决策可能会由一个轻量级LLM来驱动，以处理更模糊的情况。

3.3 实现工具执行器与护栏

工具执行器需要安全地调用函数。我们为其添加超时和基础验证：

import signal from functools import wraps from typing import Callable class TimeoutException(Exception): pass def timeout_handler(signum, frame): raise TimeoutException("Tool execution timed out") def safe_tool_executor(timeout_seconds=5): """装饰器：为工具函数添加超时和异常捕获""" def decorator(func: Callable): @wraps(func) def wrapper(*args, **kwargs): # 设置超时信号 signal.signal(signal.SIGALRM, timeout_handler) signal.alarm(timeout_seconds) try: result = func(*args, **kwargs) signal.alarm(0) # 取消闹钟 return {"success": True, "result": result} except TimeoutException: return {"success": False, "error_message": f"Tool execution exceeded {timeout_seconds} seconds"} except Exception as e: return {"success": False, "error_message": f"Tool error: {str(e)}"} finally: signal.alarm(0) # 确保总是取消闹钟 return wrapper return decorator # 应用装饰器到工具上 @safe_tool_executor(timeout_seconds=3) def query_database(sql_query: str): # 这里应连接真实数据库，此处为模拟 if "DROP TABLE" in sql_query.upper(): raise ValueError("Potentially dangerous query detected!") # 模拟查询 return [{"id": 1, "name": "Sample Data"}]

同时，我们添加一个简单的输出护栏，用于检查最终答案的格式和内容安全：

class OutputGuardrail: def validate(self, answer: str, state: AgentState) -> Dict: """验证最终输出""" issues = [] if not answer or answer.strip() == "": issues.append("Answer is empty.") if len(answer) > 1000: # 长度限制 issues.append("Answer is too long.") # 简单的内容安全检查（示例） blacklist = ["敏感词A", "内部密码"] for word in blacklist: if word in answer: issues.append(f"Answer contains inappropriate content: {word}") break if issues: return {"valid": False, "issues": issues, "sanitized_answer": "[Output blocked by guardrail]"} else: return {"valid": True, "sanitized_answer": answer}

3.4 组装主循环

现在，我们将所有组件组装到主执行循环中。这是Harness的核心驱动逻辑：

class AgentHarness: def __init__(self): self.orchestrator = RuleBasedOrchestrator() self.guardrail = OutputGuardrail() self.state = None def parse_intent_with_llm(self, user_input: str) -> str: """模拟LLM解析用户意图。实际应调用LLM API。""" # 此处为简化模拟逻辑 if "销售" in user_input and "数据" in user_input: return "query_sales_data" elif "表结构" in user_input or "字段" in user_input: return "get_table_info" else: return "general_query" def generate_plan_with_llm(self, intent: str) -> List[str]: """模拟LLM生成计划。""" plan_map = { "query_sales_data": ["get_table_schema:sales", "query_database", "explain_query_result"], "get_table_info": ["get_table_schema"], "general_query": ["need_clarification"] # 无法理解，需要澄清 } return plan_map.get(intent, ["need_clarification"]) def execute_single_step(self, step: str, state: AgentState): """执行单个步骤（工具调用或LLM生成）""" if step.startswith("get_table_schema:"): table = step.split(":")[1] result = get_table_schema(table) state.completed_steps.append({"step": step, "result": result}) elif step == "query_database": # 这里需要根据之前获取的schema构造查询，此处简化 sql = "SELECT * FROM sales LIMIT 5" # 模拟生成的SQL result = query_database(sql) state.completed_steps.append({"step": step, "result": result}) elif step == "explain_query_result": last_result = state.completed_steps[-1]["result"] if last_result["success"]: # 模拟LLM解释结果 explanation = f"查询成功，返回了{len(last_result['result'])}条记录。" state.final_answer = explanation else: state.error = "Failed to explain results." elif step == "need_clarification": state.final_answer = "抱歉，我没完全理解您的需求。您能具体说一下想查询哪些数据吗？例如‘查看上周的销售总额’。" def run(self, user_input: str) -> str: """主运行方法""" self.state = AgentState(user_input=user_input) while True: action = self.orchestrator.decide_next_action(self.state) if action == 'parse_intent': self.state.parsed_intent = self.parse_intent_with_llm(self.state.user_input) elif action == 'generate_plan': self.state.current_plan = self.generate_plan_with_llm(self.state.parsed_intent) elif action == 'execute_tool': next_step_index = len(self.state.completed_steps) if next_step_index < len(self.state.current_plan): next_step = self.state.current_plan[next_step_index] self.execute_single_step(next_step, self.state) else: self.state.error = "Plan index out of range." elif action == 'generate_final_answer': if self.state.final_answer is None: # 如果没有通过工具生成答案，则模拟LLM总结 self.state.final_answer = f"根据您的查询‘{self.state.user_input}’，已完成分析。" # 通过护栏检查 validation = self.guardrail.validate(self.state.final_answer, self.state) if validation["valid"]: return validation["sanitized_answer"] else: return f"答案生成失败：{validation['issues']}" elif action in ['error', 'finished']: return self.state.final_answer or f"处理结束，状态：{action}. 错误：{self.state.error}" else: self.state.error = f"Unknown action: {action}" return f"系统内部错误：{self.state.error}"

这个run方法体现了一个完整的感知-决策-执行-评估循环。它不断检查状态，决定下一步，执行，更新状态，直到满足终止条件（成功、失败或超限）。

4. 关键问题：如何设计有效的评估与迭代循环？

构建Harness不是一劳永逸的，产品级Agent必须能持续改进。这就需要建立闭环的评估与迭代机制。我们的原型中，评估是隐式的（通过规则判断成功/失败）。但在真实产品中，你需要更系统的评估体系。

4.1 多维度评估指标

不能只用一个“准确率”来衡量Agent。一个产品级Agent需要从多个维度评估：

在产品初期，可以优先关注任务完成率和异常退出率。一个连基本流程都走不通的Agent，其他指标再好也无意义。

4.2 构建评估工作流

评估不应是手动的。你需要一个自动化的评估工作流：

1. 测试集管理：维护一个覆盖核心场景、边界案例和对抗性输入的测试用例库。每个用例包括输入、预期输出和允许的工具调用序列。
2. 自动化运行：定期（如每夜）或在新模型/代码发布后，用测试集全量运行你的Harness。
3. 自动评分：根据评估维度，对每次运行的结果进行自动评分。功能性指标可以通过规则或模型打分；性能指标直接从监控数据获取。
4. 结果分析与归因：当评分下降时，需要快速定位原因。是LLM理解错了？还是工具调用出错了？或是护栏误杀了？这需要Harness提供详细的**执行轨迹（Trace）**日志。

一个完整的Trace日志应该像飞机黑匣子，记录每个环节的输入输出：

{ "session_id": "abc123", "user_input": "帮我查一下上个月的销售冠军", "steps": [ { "step_id": 1, "action": "intent_parsing", "input": "帮我查一下上个月的销售冠军", "output": {"intent": "query_top_salesperson", "period": "last_month"}, "timestamp": "2023-10-27T10:00:00Z", "latency_ms": 450, "llm_usage": {"prompt_tokens": 56, "completion_tokens": 12} }, { "step_id": 2, "action": "tool_call", "tool_name": "query_database", "parameters": {"sql": "SELECT salesperson_id FROM sales WHERE date >= '2023-09-01' GROUP BY ..."}, "result": {"success": true, "data": [...]}, "error": null, "timestamp": "2023-10-27T10:00:01Z", "latency_ms": 120 } // ... 更多步骤 ], "final_output": "上个月的销售冠军是张三，总销售额为50万元。", "guardrail_checks_passed": true, "total_latency_ms": 2100, "total_token_usage": 345 }

这样的Trace是进行问题诊断和效果优化的黄金数据。

4.3 基于评估的迭代策略

拿到评估结果和Trace后，如何改进？

• LLM层面：如果问题出在意图解析或计划生成不准，可以考虑：1) 优化Prompt（增加示例、更清晰的指令）；2) 对特定任务进行微调（Fine-tuning）；3) 切换到更适合该任务的基础模型。
• 工具层面：如果工具调用经常失败或返回错误数据，需要：1) 增强工具的健壮性和错误处理；2) 改进工具的描述（Tool Description），让LLM更准确地理解其功能和使用方式；3) 增加更多的输入验证。
• 编排逻辑层面：如果Agent容易陷入死循环或做出错误决策，需要：1) 优化Orchestrator的决策规则或模型；2) 引入更强大的评估器（Critic）在每一步后评估结果的好坏，决定继续还是回退。
• 护栏层面：如果护栏漏掉了有害输出，需要扩充过滤词库和检测规则；如果护栏误杀太多，则需要调整其敏感度，或采用更精细的基于模型的分类器。

这个“运行 -> 评估 -> 分析 -> 优化”的循环，是产品级Agent能够持续进化的生命线。

5. 生产环境部署与监控考量

将原型Harness部署到生产环境，会面临一系列新的挑战。

5.1 可观测性（Observability）建设

“黑盒”AI系统是运维的噩梦。你必须建立三大支柱：

• 日志（Logging）：除了上文提到的结构化执行Trace，还需要记录所有LLM API调用（请求/响应）、工具调用、护栏决策等，并统一收集到如ELK或Loki这样的日志系统中，便于搜索和聚合分析。
• 指标（Metrics）：定义并暴露关键业务和技术指标。例如：
  • agent_requests_total：总请求数。
  • agent_success_rate：任务成功完成率。
  • agent_latency_seconds：请求延迟分布。
  • llm_token_usage：Token消耗的统计。
  • tool_failure_count：各工具调用失败次数。 这些指标应接入Prometheus等监控系统，并设置告警（如成功率低于95%时触发）。
• 追踪（Tracing）：对于一个用户请求在Harness内部流经多个服务（LLM API、数据库、内部微服务）的复杂情况，需要分布式追踪（如Jaeger）来可视化整个调用链，精准定位延迟瓶颈。

5.2 弹性与容错设计

• 重试与降级：LLM API调用可能因网络或服务方原因失败。必须实现带退避策略的智能重试（如指数退避）。对于非核心步骤，在多次重试失败后应有降级方案（例如，无法生成图文并茂的报告时，至少返回文本摘要）。
• 限流与熔断：防止上游LLM服务过载或自身被突发流量打垮。需要实现请求限流（Rate Limiting）。当检测到下游服务（如某个工具或LLM API）失败率过高时，应自动熔断（Circuit Breaker），快速失败并返回友好提示，避免资源耗尽。
• 状态持久化：对于长会话或复杂任务，Agent的状态必须能持久化到数据库（如Redis或PostgreSQL）。这样，即使服务实例重启，用户也能从中断处继续，保障体验的连续性。

5.3 成本控制与优化

LLM API调用是主要成本中心。必须精细化管理：

• 缓存策略：对于频繁出现的、结果确定的用户查询（如“公司的退货政策是什么？”），可以将LLM的最终答案或中间表示（如向量嵌入）缓存起来，直接返回，避免重复计算。
• 模型路由：并非所有任务都需要最强大、最昂贵的模型（如GPT-4）。可以建立一个路由层，根据任务的复杂度（可通过首次意图解析判断），将其分配给不同能力的模型（如简单QA用GPT-3.5-Turbo，复杂推理用GPT-4）。这需要在效果和成本间取得平衡。
• Token使用分析：定期分析日志，找出Prompt过长或Completion冗余的环节。优化Prompt设计，减少不必要的上下文，使用系统消息（System Message）更有效地约束模型行为，都是降低Token消耗的有效手段。

6. 避坑指南与经验总结

在从零搭建产品级Agent Harness的过程中，我踩过不少坑，也积累了一些关键心得。

核心心得：先做“笨”的确定性系统，再逐步引入“聪明”的不确定性。很多团队一开始就追求全LLM驱动的、高度灵活的智能体，结果陷入调试地狱。更好的路径是：先用规则和模板实现核心流程的80%，确保它稳定、可控、可调试。然后，在关键且风险可控的环节（如意图分类、答案润色）引入LLM，用其能力提升体验。这样，系统的主体骨架是坚实的，AI只是增强肌肉，而不是充当随时可能散架的骨骼。

避坑点1：过度依赖LLM的规划能力让LLM自由规划多步任务（Plan）听起来很美好，但在生产环境中极易失控。LLM可能会生成不存在的工具调用、陷入循环或产生不安全的步骤。我们的策略是约束性规划：预先定义好几种标准的任务流程模板（Workflow Template），LLM的工作只是将用户输入匹配到最合适的模板，并填充模板中的参数。这大大降低了复杂性和风险。

避坑点2：忽视工具执行的副作用工具调用可能修改数据库、发送邮件、调用外部API。必须实施最小权限原则和模拟执行模式。在开发测试阶段，所有写操作的工具都应先接入“模拟器”，只记录而不真实执行。上线前，必须对每个工具的副作用进行严格评审。对于高风险操作（如删除、支付），应在流程中内置人工确认环节或二次授权。

避坑点3：评估体系与业务目标脱节不要为了评估而评估。你优化的指标必须与最终的业务目标对齐。如果业务目标是提升客服效率，那么“首次对话解决率”和“平均处理时间”就比“答案的BLEU分数”更重要。在构建评估集时，必须与业务方紧密合作，确保测试用例真实反映用户场景和成功标准。

避坑点4：忽略“沉默的失败”Agent没有报错，但给出了一个完全错误的答案，这是最危险的情况。除了输出护栏，还需要建立端到端的集成测试和线上巡检机制。定期用一批已知答案的“哨兵问题”对生产环境进行测试，监控其答案质量的变化。一旦发现漂移，立即告警。

构建产品级Agent Harness是一个典型的系统工程，它要求我们在对AI能力保持热情的同时，对软件工程的严谨性抱有最高的敬畏。它不是一次性的开发，而是一个需要持续观察、测量、调整和演进的有机体。从这个原型出发，你可以根据实际业务需求，逐步强化它的每一个模块——更智能的编排器、更丰富的工具库、更坚固的护栏、更高效的评估循环，最终让它成为你产品中可靠且强大的智能核心。