最近在尝试将AI智能体Agent落地到实际业务场景时发现一个普遍痛点很多教程只讲单次对话或简单工具调用一旦涉及复杂、多步骤的自动化任务代码就变得难以维护和迭代。Agent的“智能”往往停留在演示阶段无法形成稳定可靠的工程化能力。这正是“循环工程”Cyclical Engineering与“Harness工程”要解决的核心问题。本文将深入探讨如何为AI智能体构建一个健壮、可观测、可迭代的工程化架构。我们将从核心概念入手逐步拆解一个基于流行框架如LangChain/LangGraph的实战项目涵盖从架构设计、代码实现到部署监控的全流程。无论你是希望将AI能力集成到现有系统的开发者还是正在探索智能体落地的工程师都能从中获得一套可直接复用的方法论和代码模板。1. 理解核心概念循环工程与Harness工程在深入代码之前我们必须厘清两个关键概念“循环工程”与“Harness工程”。它们并非某个特定框架的名称而是一种构建复杂AI智能体的系统化设计思想。1.1 什么是循环工程Cyclical Engineering循环工程指的是智能体执行任务时的一种核心运行模式——感知、思考、行动、观察的循环。这与人类解决问题的方式类似感知Perceive获取环境信息用户输入、API返回、数据库状态。思考Think分析信息规划下一步行动调用哪个工具如何组合。行动Act执行规划好的动作调用API、运行代码、查询知识库。观察Observe评估行动结果更新内部状态。这个循环会持续进行直到任务达成或无法继续。循环工程强调对这个过程的显式建模、状态管理和流程控制确保智能体的行为是可预测、可调试的。1.2 什么是Harness工程“Harness”原意为“马具”或“安全带”在工程中引申为“约束、控制和管理系统”。Harness工程指的是为AI智能体构建一套安全、可控、可观测的“防护栏”和“驾驶舱”。它主要包含以下几个层面安全与合规护栏Safety Guardrails在智能体行动前后进行检查防止其执行危险、不道德或超出权限的操作如删除生产数据、访问未授权API。状态管理与持久化State Management可靠地保存和恢复智能体在长周期对话或多步骤任务中的中间状态。可观测性与监控Observability记录智能体的决策链路、工具调用、token消耗、执行耗时等便于问题排查和性能优化。流程编排与调度Orchestration管理多个智能体或工具之间的协作关系定义复杂的执行流程图。1.3 为什么需要这种工程化架构没有工程化架构的智能体就像一辆没有方向盘和刹车的跑车速度可能很快但方向失控极其危险。具体问题包括状态丢失对话或任务中断后无法恢复。行为不可控可能执行未授权的操作。调试困难出错时不知道是哪个环节、哪次调用出了问题。难以迭代业务逻辑、工具和模型升级时代码牵一发而动全身。循环工程与Harness工程相结合旨在将AI智能体从“演示玩具”升级为“生产级组件”。2. 环境准备与核心工具选型在开始实战前我们需要搭建开发环境并选择合适的技术栈。本文将以Python生态为例使用目前最流行的Agent开发框架之一。2.1 环境与版本说明操作系统Windows 10/11, macOS, 或 Linux (Ubuntu 20.04)。本文命令以Linux/macOS为例。Python版本 3.10 或 3.11。这是大多数AI框架稳定支持的版本。包管理工具pip或poetry。推荐使用poetry管理依赖和虚拟环境。核心框架langchain和langgraph。LangChain提供了构建链和智能体的基础组件而LangGraph专门用于创建有状态、多步骤的循环工作流。大语言模型LLM为了本地化和快速演示我们将使用Ollama本地运行开源模型如llama3.1、qwen2.5。你也可以替换为OpenAI、Anthropic等云端API。向量数据库可选用于为智能体增加知识库RAG能力。本文使用轻量级的ChromaDB。重要提示以下版本号可能会快速迭代请根据官方文档调整。本文重点在于架构思路代码具有通用性。2.2 初始化项目与安装依赖首先创建一个新的项目目录并初始化虚拟环境。# 创建项目目录 mkdir ai-agent-harness cd ai-agent-harness # 使用 poetry 初始化项目 (如果没有poetry请先安装: pip install poetry) poetry init -n poetry add langchain langgraph langchain-community ollama chromadb poetry add pydantic python-dotenv # 用于配置管理和数据验证 # 或者使用 pip python -m venv venv source venv/bin/activate # Windows: venv\Scripts\activate pip install langchain langgraph langchain-community ollama chromadb pydantic python-dotenv接下来确保Ollama服务已安装并运行并拉取一个模型。# 安装Ollama (请参考官网: https://ollama.com) # 拉取一个模型例如 Llama 3.1 ollama pull llama3.1:8b项目基础结构如下ai-agent-harness/ ├── pyproject.toml/poetry.lock # 依赖管理 ├── .env # 环境变量如API密钥 ├── src/ │ ├── __init__.py │ ├── agents/ # 智能体定义 │ │ ├── __init__.py │ │ └── planner_agent.py │ ├── tools/ # 自定义工具 │ │ ├── __init__.py │ │ └── calculator.py │ ├── graphs/ # 工作流图定义核心 │ │ ├── __init__.py │ │ └── research_agent_graph.py │ ├── state/ # 状态模式定义 │ │ ├── __init__.py │ │ └── agent_state.py │ ├── harness/ # 安全护栏与监控 │ │ ├── __init__.py │ │ ├── safety_check.py │ │ └── observability.py │ └── main.py # 应用入口 └── tests/3. 架构核心定义智能体状态与工作流图循环工程的核心是管理状态。我们使用Pydantic来定义智能体在整个执行周期中所处的状态。3.1 定义状态模式State Schema在src/state/agent_state.py中我们定义一个名为AgentState的类它继承自TypedDict用于描述工作流中流动的数据。# src/state/agent_state.py from typing import TypedDict, List, Annotated from typing_extensions import TypedDict import operator # 使用LangGraph的注解来定义状态的聚合方式 class AgentState(TypedDict): 智能体工作流的全局状态容器。 # 用户输入的问题或目标 input: str # 智能体“思考”后生成的计划步骤列表 plan: List[str] # 到目前为止已执行完成的步骤列表 completed_steps: List[str] # 当前步骤的执行结果或观察到的信息 observation: str # 从开始到现在的完整执行历史用于最终报告或调试 history: Annotated[List[str], operator.add] # 关键这个字段会自动累加 # 最终答案或输出 output: str关键点解释Annotated[List[str], operator.add]这是LangGraph的一个强大特性。它声明history字段是一个列表当工作流中多个节点修改这个字段时它们的修改即向列表中添加元素会自动合并add操作而不是覆盖。这极大简化了状态管理。状态字段根据你的智能体任务自定义。例如可以添加knowledge_base检索到的文档、tools_called调用过的工具列表等。3.2 构建基础工作流图Graph工作流图由“节点”Node和“边”Edge组成。节点是执行单元函数边定义了节点之间的流转条件。我们在src/graphs/research_agent_graph.py中创建一个简单的“研究助理”智能体图。# src/graphs/research_agent_graph.py from langgraph.graph import StateGraph, END from .state.agent_state import AgentState from langchain_community.chat_models import ChatOllama from langchain_core.messages import HumanMessage # 初始化一个本地LLM llm ChatOllama(modelllama3.1:8b, temperature0) # 1. 定义节点函数 def planner_node(state: AgentState) - dict: 规划节点分析输入拆解任务步骤。 print(f[Planner] 正在规划任务: {state[input]}) # 让LLM生成一个步骤计划 prompt f 用户的目标是{state[input]} 请将这个目标拆解成3-5个具体的、可执行的步骤。 以清晰的列表形式返回每个步骤一行。 message [HumanMessage(contentprompt)] response llm.invoke(message) plan response.content.strip().split(\n) # 更新状态 return {plan: plan, history: [f规划器生成了计划: {plan}]} def executor_node(state: AgentState) - dict: 执行节点执行当前计划中的第一个未完成步骤。 # 找出第一个未完成的步骤 all_steps state[plan] completed state.get(completed_steps, []) remaining [s for s in all_steps if s not in completed] if not remaining: return {observation: 所有步骤已完成。, output: 任务执行完毕。} current_step remaining[0] print(f[Executor] 正在执行步骤: {current_step}) # 模拟执行步骤这里可以替换为真正的工具调用如网络搜索、代码执行等 # 例如我们让LLM模拟执行并给出结果 prompt f 模拟执行以下研究步骤并生成一段简短的发现或结果 步骤{current_step} message [HumanMessage(contentprompt)] response llm.invoke(message) observation response.content.strip() # 更新状态标记该步骤完成记录观察结果 new_completed completed [current_step] return { observation: observation, completed_steps: new_completed, history: [f执行了步骤 {current_step}结果: {observation}] } def reporter_node(state: AgentState) - dict: 报告节点汇总所有观察结果生成最终报告。 print(f[Reporter] 正在生成最终报告。历史记录: {state[history]}) prompt f 基于以下执行历史和观察结果为用户生成一份完整、结构化的总结报告 初始问题{state[input]} 执行历史 {chr(10).join(state[history])} message [HumanMessage(contentprompt)] response llm.invoke(message) final_report response.content.strip() return {output: final_report, history: [f报告生成完毕: {final_report[:100]}...]} # 2. 构建图 def create_research_agent_graph(): 创建并返回一个研究助理的工作流图。 workflow StateGraph(AgentState) # 添加节点 workflow.add_node(planner, planner_node) workflow.add_node(executor, executor_node) workflow.add_node(reporter, reporter_node) # 设置入口点 workflow.set_entry_point(planner) # 添加边定义流程逻辑 workflow.add_edge(planner, executor) # 从执行器出来需要判断是否还有步骤未完成 workflow.add_conditional_edges( executor, # 条件判断函数如果所有步骤完成前往reporter否则继续执行 lambda state: reporter if len(state.get(completed_steps, [])) len(state.get(plan, [])) else executor, { executor: executor, # 继续循环执行 reporter: reporter # 去生成报告 } ) workflow.add_edge(reporter, END) # 编译图 return workflow.compile() # 快捷方式 research_graph create_research_agent_graph()这个图实现了基本的“规划-执行-报告”循环planner节点先规划步骤。executor节点循环执行每个步骤直到所有步骤标记为完成。reporter节点汇总生成最终报告。4. 实战深化集成工具、安全护栏与可观测性仅有循环不够我们需要Harness工程来加固它。4.1 创建自定义工具Tools工具是智能体与外界交互的“手”。我们创建一个简单的计算器工具并展示如何集成到智能体中。# src/tools/calculator.py from langchain.tools import tool from pydantic import BaseModel, Field # 定义工具的输入模式这能帮助LLM正确生成调用参数 class CalculatorInput(BaseModel): a: float Field(description第一个数字) b: float Field(description第二个数字) operator: str Field(description运算符支持 add, subtract, multiply, divide) tool(args_schemaCalculatorInput) def calculator_tool(a: float, b: float, operator: str) - str: 执行简单的数学计算。 try: if operator add: result a b elif operator subtract: result a - b elif operator multiply: result a * b elif operator divide: if b 0: return 错误除数不能为零。 result a / b else: return f错误不支持的运算符 {operator}。支持: add, subtract, multiply, divide return f计算结果: {a} {operator} {b} {result} except Exception as e: return f计算过程中发生错误: {e}4.2 实现安全护栏Safety Guardrails安全护栏在动作执行前后进行检查。我们实现一个简单的“权限检查”和“输出过滤”护栏。# src/harness/safety_check.py from typing import Dict, Any import re class SafetyHarness: 安全护栏示例。 staticmethod def pre_execution_guard(state: Dict[str, Any], intended_action: str) - tuple[bool, str]: 在执行任何工具或操作前检查。 # 示例1检查是否试图执行危险命令 dangerous_patterns [rrm\s-rf, rformat\sc:, rdrop\sdatabase, rsudo] for pattern in dangerous_patterns: if re.search(pattern, intended_action, re.IGNORECASE): return False, f安全拦截检测到潜在危险操作 {intended_action} # 示例2检查输入是否包含敏感信息简单演示 if password in intended_action or 密钥 in intended_action: return False, 安全拦截输入中可能包含敏感词汇。 print(f[Safety Guard] 预检通过: {intended_action[:50]}...) return True, 预检通过 staticmethod def post_execution_guard(observation: str) - tuple[bool, str]: 在获得工具执行结果后检查。 # 示例过滤结果中的个人身份信息PII - 简单正则示例 # 在实际项目中应使用更专业的PII检测库 pii_patterns { r\b\d{18}\b: [身份证号已屏蔽], r\b1[3-9]\d{9}\b: [手机号已屏蔽], r\b[A-Za-z0-9._%-][A-Za-z0-9.-]\.[A-Z|a-z]{2,}\b: [邮箱已屏蔽], } filtered_obs observation for pattern, replacement in pii_patterns.items(): filtered_obs re.sub(pattern, replacement, filtered_obs) if filtered_obs ! observation: print(f[Safety Guard] 输出内容已进行PII过滤。) return True, filtered_obs # 返回过滤后的结果 return True, observation # 原样返回4.3 集成工具与护栏到工作流节点我们修改之前的executor_node使其能够动态选择并安全地调用工具。# 在 research_agent_graph.py 中更新或新建一个节点 from src.tools.calculator import calculator_tool from src.harness.safety_check import SafetyHarness def enhanced_executor_node(state: AgentState) - dict: 增强的执行节点能调用工具并经过安全护栏检查。 all_steps state[plan] completed state.get(completed_steps, []) remaining [s for s in all_steps if s not in completed] if not remaining: return {observation: 所有步骤已完成。, output: 任务执行完毕。} current_step remaining[0] print(f[Enhanced Executor] 处理步骤: {current_step}) # 1. 预执行安全检查 safe_to_run, pre_check_msg SafetyHarness.pre_execution_guard(state, current_step) if not safe_to_run: return {observation: f执行被安全护栏阻止: {pre_check_msg}, completed_steps: completed [current_step]} # 2. 判断步骤类型并选择行动 observation # 简单启发式如果步骤描述中包含“计算”或数字尝试使用计算器工具 if 计算 in current_step or any(char.isdigit() for char in current_step): # 这里简化处理实际中应该用LLM来解析步骤并生成工具调用参数 # 我们假设步骤是类似“计算一下 123 乘以 456 的结果” try: # 非常简单的提取数字逻辑仅用于演示 import re numbers re.findall(r\d, current_step) if len(numbers) 2: a, b int(numbers[0]), int(numbers[1]) # 调用工具 tool_result calculator_tool.invoke({a: a, b: b, operator: multiply}) observation f使用计算器工具得到: {tool_result} else: observation f无法从步骤{current_step}中提取足够数字进行计算。 except Exception as e: observation f调用计算工具时出错: {e} else: # 其他步骤仍用LLM模拟 prompt f模拟执行研究步骤{current_step} message [HumanMessage(contentprompt)] response llm.invoke(message) observation response.content.strip() # 3. 后执行安全检查与过滤 _, filtered_observation SafetyHarness.post_execution_guard(observation) # 更新状态 new_completed completed [current_step] return { observation: filtered_observation, completed_steps: new_completed, history: [f执行了步骤 {current_step}结果: {filtered_observation[:100]}...] }4.4 添加可观测性日志与监控可观测性让我们能洞察智能体的内部运行。我们实现一个简单的装饰器来记录节点的执行情况。# src/harness/observability.py import time import functools from datetime import datetime class Observability: 可观测性模块用于记录节点执行跟踪。 logs [] staticmethod def log_node_invocation(node_name: str, state_input: dict, duration: float, result: dict): log_entry { timestamp: datetime.utcnow().isoformat(), node: node_name, input_state_snapshot: {k: str(v)[:200] for k, v in state_input.items()}, # 截断长内容 execution_duration_ms: round(duration * 1000, 2), result_state_update: {k: str(v)[:200] for k, v in result.items()}, } Observability.logs.append(log_entry) print(f[Observability] {node_name} 执行耗时: {log_entry[execution_duration_ms]}ms) # 在实际项目中这里可以将日志发送到ELK、Prometheus等系统 staticmethod def track_node(func): 装饰器用于自动跟踪节点函数的执行。 functools.wraps(func) def wrapper(state): start_time time.time() result func(state) end_time time.time() Observability.log_node_invocation(func.__name__, state, end_time - start_time, result) return result return wrapper # 使用装饰器增强节点函数 # 只需在节点函数定义前加上 Observability.track_node # 例如 # Observability.track_node # def planner_node(state: AgentState) - dict: # ...5. 完整运行与测试现在我们将所有部分组合起来创建一个完整的、具备Harness能力的智能体应用。# src/main.py from src.graphs.research_agent_graph import create_research_agent_graph from src.harness.observability import Observability import asyncio async def main(): print( 启动具备Harness的AI研究智能体 ) # 1. 获取编译好的工作流图 # 注意在实际的research_agent_graph.py中我们需要用装饰器包装节点函数 # 这里假设我们已经用 Observability.track_node 装饰了 planner_node, enhanced_executor_node, reporter_node app create_research_agent_graph() # 2. 定义初始状态 initial_state { input: 请研究一下太阳能光伏发电的基本原理并计算如果一块标准光伏板效率20%在日照强度为1000W/m²下工作5小时能产生多少度电假设光伏板面积是2平方米。, plan: [], completed_steps: [], observation: , history: [], output: } # 3. 运行图 print(f\n用户输入: {initial_state[input]}) print(- * 50) final_state None # LangGraph编译的app是可调用对象我们也可以流式获取每个步骤 async for step in app.astream(initial_state, stream_modevalues): node_name list(step.keys())[0] if step else unknown print(f[Graph Step] 节点 {node_name} 执行完毕。) # 最后一个状态就是最终状态 final_state step.get(node_name, {}) print(- * 50) print(\n 任务执行完成 ) print(f最终输出:\n{final_state.get(output, 无输出)}) # 4. 打印可观测性日志 print(\n 执行跟踪日志 ) for i, log in enumerate(Observability.logs): print(f{i1}. [{log[timestamp]}] {log[node]} - {log[execution_duration_ms]}ms) if __name__ __main__: asyncio.run(main())运行这个程序你将看到智能体按照“规划-循环执行-报告”的流程工作并且每一步都有日志输出和安全检查如果触发。这便是一个具备基本循环工程和Harness工程特性的AI智能体。6. 常见问题与排查思路在构建和运行此类智能体系统时你可能会遇到以下典型问题问题现象可能原因排查思路与解决方案图编译失败StateGraph状态模式定义错误节点函数返回值与状态字段不匹配。1. 检查AgentState的TypedDict定义确保字段名拼写正确。2. 确保每个节点函数返回一个字典其键必须是AgentState中定义的字段的子集。智能体陷入无限循环条件边add_conditional_edges的逻辑判断函数有误未能正确终止。1. 在条件函数中打印state进行调试。2. 确保终止条件如len(completed_steps) len(plan)逻辑严密。可以在循环超过N次后强制跳出。工具调用失败或参数错误LLM生成的工具调用参数格式不符合工具期望的args_schema。1. 为工具提供清晰、结构化的描述和参数定义。2. 使用LangChain的StructuredTool或Tool的args_schema特性。3. 在调用工具前可以增加一个“参数验证与修正”节点。状态History未正确累积未使用Annotated[List[str], operator.add]注解来声明可累加字段。确保在状态模式中希望被多个节点追加内容的字段如history使用Annotated和operator.add进行注解。这是LangGraph实现状态聚合的关键。Ollama模型调用超时或无响应Ollama服务未启动模型未正确下载网络问题。1. 运行ollama serve确保服务在运行。2. 运行ollama list确认模型已存在。3. 在代码中尝试使用timeout参数并捕获异常。安全护栏误拦截正常请求正则表达式或规则过于严格。1. 细化规则逻辑区分上下文。2. 实现护栏的“学习模式”或“审核模式”将拦截记录交由人工复审逐步优化规则。执行历史过长导致性能下降history或state随着循环次数增加变得巨大。1. 定期对历史进行摘要Summarize只保留关键信息。2. 将历史存储到外部数据库如Redis状态中只保留引用ID。7. 最佳实践与工程建议将AI智能体投入生产环境需要遵循更严格的软件工程准则。状态设计要精简且聚焦只将真正需要跨节点共享和修改的数据放入State。对于大型数据如检索到的文档全文在状态中存储引用如ID或索引而非数据本身。使用Pydantic进行严格的输入输出验证确保数据类型安全。工具设计遵循单一职责原则每个工具应只做一件事并做好错误处理。工具的描述description要极其精确这直接影响到LLM能否正确调用它。为工具编写单元测试确保其功能稳定。实现分级安全策略基础层语法/规则如本文所示的正则表达式过滤。模型层语义使用一个轻量级的“审查模型”对智能体的计划和输出进行二次评分和过滤。人工层关键操作对于涉及资金、数据删除、对外发布等高风险操作设计“人工审批”节点将操作挂起等待人工确认。建立全面的可观测体系日志记录每个节点的输入/输出、耗时、Token使用量。链路追踪为每次用户会话分配唯一trace_id串联所有相关日志和调用。指标监控监控智能体的平均完成时间、成功率、工具调用分布、错误类型。成本监控密切监控不同模型和工具的调用成本。版本化与回滚将智能体的工作流图Graph、工具集、提示词模板进行版本控制如Git。部署新版本时可以采用蓝绿部署或金丝雀发布策略先让小部分流量走新版本。必须保留快速回滚到上一稳定版本的能力。测试策略单元测试测试每个独立的工具和节点函数。集成测试测试整个工作流图使用固定的输入验证输出是否符合预期。模糊测试/对抗测试输入一些边缘案例、无意义或带有诱导性的问题观察智能体行为是否安全、稳定。通过以上架构和最佳实践你可以构建出不仅“智能”而且“可靠、安全、可维护”的AI智能体系统。这标志着AI应用从原型走向生产的关键一步。 最近在尝试将AI智能体Agent落地到实际业务场景时发现一个普遍痛点很多教程只讲单次对话或简单工具调用一旦涉及复杂、多步骤的自动化任务代码就变得难以维护和迭代。Agent的“智能”往往停留在演示阶段无法形成稳定可靠的工程化能力。这正是“循环工程”Cyclical Engineering与“Harness工程”要解决的核心问题。本文将深入探讨如何为AI智能体构建一个健壮、可观测、可迭代的工程化架构。我们将从核心概念入手逐步拆解一个基于流行框架如LangChain/LangGraph的实战项目涵盖从架构设计、代码实现到部署监控的全流程。无论你是希望将AI能力集成到现有系统的开发者还是正在探索智能体落地的工程师都能从中获得一套可直接复用的方法论和代码模板。1. 理解核心概念循环工程与Harness工程在深入代码之前我们必须厘清两个关键概念“循环工程”与“Harness工程”。它们并非某个特定框架的名称而是一种构建复杂AI智能体的系统化设计思想。1.1 什么是循环工程Cyclical Engineering循环工程指的是智能体执行任务时的一种核心运行模式——感知、思考、行动、观察的循环。这与人类解决问题的方式类似感知Perceive获取环境信息用户输入、API返回、数据库状态。思考Think分析信息规划下一步行动调用哪个工具如何组合。行动Act执行规划好的动作调用API、运行代码、查询知识库。观察Observe评估行动结果更新内部状态。这个循环会持续进行直到任务达成或无法继续。循环工程强调对这个过程的显式建模、状态管理和流程控制确保智能体的行为是可预测、可调试的。1.2 什么是Harness工程“Harness”原意为“马具”或“安全带”在工程中引申为“约束、控制和管理系统”。Harness工程指的是为AI智能体构建一套安全、可控、可观测的“防护栏”和“驾驶舱”。它主要包含以下几个层面安全与合规护栏Safety Guardrails在智能体行动前后进行检查防止其执行危险、不道德或超出权限的操作如删除生产数据、访问未授权API。状态管理与持久化State Management可靠地保存和恢复智能体在长周期对话或多步骤任务中的中间状态。可观测性与监控Observability记录智能体的决策链路、工具调用、token消耗、执行耗时等便于问题排查和性能优化。流程编排与调度Orchestration管理多个智能体或工具之间的协作关系定义复杂的执行流程图。1.3 为什么需要这种工程化架构没有工程化架构的智能体就像一辆没有方向盘和刹车的跑车速度可能很快但方向失控极其危险。具体问题包括状态丢失对话或任务中断后无法恢复。行为不可控可能执行未授权的操作。调试困难出错时不知道是哪个环节、哪次调用出了问题。难以迭代业务逻辑、工具和模型升级时代码牵一发而动全身。循环工程与Harness工程相结合旨在将AI智能体从“演示玩具”升级为“生产级组件”。2. 环境准备与核心工具选型在开始实战前我们需要搭建开发环境并选择合适的技术栈。本文将以Python生态为例使用目前最流行的Agent开发框架之一。2.1 环境与版本说明操作系统Windows 10/11, macOS, 或 Linux (Ubuntu 20.04)。本文命令以Linux/macOS为例。Python版本 3.10 或 3.11。这是大多数AI框架稳定支持的版本。包管理工具pip或poetry。推荐使用poetry管理依赖和虚拟环境。核心框架langchain和langgraph。LangChain提供了构建链和智能体的基础组件而LangGraph专门用于创建有状态、多步骤的循环工作流。大语言模型LLM为了本地化和快速演示我们将使用Ollama本地运行开源模型如llama3.1、qwen2.5。你也可以替换为OpenAI、Anthropic等云端API。向量数据库可选用于为智能体增加知识库RAG能力。本文使用轻量级的ChromaDB。重要提示以下版本号可能会快速迭代请根据官方文档调整。本文重点在于架构思路代码具有通用性。2.2 初始化项目与安装依赖首先创建一个新的项目目录并初始化虚拟环境。# 创建项目目录 mkdir ai-agent-harness cd ai-agent-harness # 使用 poetry 初始化项目 (如果没有poetry请先安装: pip install poetry) poetry init -n poetry add langchain langgraph langchain-community ollama chromadb poetry add pydantic python-dotenv # 用于配置管理和数据验证 # 或者使用 pip python -m venv venv source venv/bin/activate # Windows: venv\Scripts\activate pip install langchain langgraph langchain-community ollama chromadb pydantic python-dotenv接下来确保Ollama服务已安装并运行并拉取一个模型。# 安装Ollama (请参考官网: https://ollama.com) # 拉取一个模型例如 Llama 3.1 ollama pull llama3.1:8b项目基础结构如下ai-agent-harness/ ├── pyproject.toml/poetry.lock # 依赖管理 ├── .env # 环境变量如API密钥 ├── src/ │ ├── __init__.py │ ├── agents/ # 智能体定义 │ │ ├── __init__.py │ │ └── planner_agent.py │ ├── tools/ # 自定义工具 │ │ ├── __init__.py │ │ └── calculator.py │ ├── graphs/ # 工作流图定义核心 │ │ ├── __init__.py │ │ └── research_agent_graph.py │ ├── state/ # 状态模式定义 │ │ ├── __init__.py │ │ └── agent_state.py │ ├── harness/ # 安全护栏与监控 │ │ ├── __init__.py │ │ ├── safety_check.py │ │ └── observability.py │ └── main.py # 应用入口 └── tests/3. 架构核心定义智能体状态与工作流图循环工程的核心是管理状态。我们使用Pydantic来定义智能体在整个执行周期中所处的状态。3.1 定义状态模式State Schema在src/state/agent_state.py中我们定义一个名为AgentState的类它继承自TypedDict用于描述工作流中流动的数据。# src/state/agent_state.py from typing import TypedDict, List, Annotated from typing_extensions import TypedDict import operator # 使用LangGraph的注解来定义状态的聚合方式 class AgentState(TypedDict): 智能体工作流的全局状态容器。 # 用户输入的问题或目标 input: str # 智能体“思考”后生成的计划步骤列表 plan: List[str] # 到目前为止已执行完成的步骤列表 completed_steps: List[str] # 当前步骤的执行结果或观察到的信息 observation: str # 从开始到现在的完整执行历史用于最终报告或调试 history: Annotated[List[str], operator.add] # 关键这个字段会自动累加 # 最终答案或输出 output: str关键点解释Annotated[List[str], operator.add]这是LangGraph的一个强大特性。它声明history字段是一个列表当工作流中多个节点修改这个字段时它们的修改即向列表中添加元素会自动合并add操作而不是覆盖。这极大简化了状态管理。状态字段根据你的智能体任务自定义。例如可以添加knowledge_base检索到的文档、tools_called调用过的工具列表等。3.2 构建基础工作流图Graph工作流图由“节点”Node和“边”Edge组成。节点是执行单元函数边定义了节点之间的流转条件。我们在src/graphs/research_agent_graph.py中创建一个简单的“研究助理”智能体图。# src/graphs/research_agent_graph.py from langgraph.graph import StateGraph, END from .state.agent_state import AgentState from langchain_community.chat_models import ChatOllama from langchain_core.messages import HumanMessage # 初始化一个本地LLM llm ChatOllama(modelllama3.1:8b, temperature0) # 1. 定义节点函数 def planner_node(state: AgentState) - dict: 规划节点分析输入拆解任务步骤。 print(f[Planner] 正在规划任务: {state[input]}) # 让LLM生成一个步骤计划 prompt f 用户的目标是{state[input]} 请将这个目标拆解成3-5个具体的、可执行的步骤。 以清晰的列表形式返回每个步骤一行。 message [HumanMessage(contentprompt)] response llm.invoke(message) plan response.content.strip().split(\n) # 更新状态 return {plan: plan, history: [f规划器生成了计划: {plan}]} def executor_node(state: AgentState) - dict: 执行节点执行当前计划中的第一个未完成步骤。 # 找出第一个未完成的步骤 all_steps state[plan] completed state.get(completed_steps, []) remaining [s for s in all_steps if s not in completed] if not remaining: return {observation: 所有步骤已完成。, output: 任务执行完毕。} current_step remaining[0] print(f[Executor] 正在执行步骤: {current_step}) # 模拟执行步骤这里可以替换为真正的工具调用如网络搜索、代码执行等 # 例如我们让LLM模拟执行并给出结果 prompt f 模拟执行以下研究步骤并生成一段简短的发现或结果 步骤{current_step} message [HumanMessage(contentprompt)] response llm.invoke(message) observation response.content.strip() # 更新状态标记该步骤完成记录观察结果 new_completed completed [current_step] return { observation: observation, completed_steps: new_completed, history: [f执行了步骤 {current_step}结果: {observation}] } def reporter_node(state: AgentState) - dict: 报告节点汇总所有观察结果生成最终报告。 print(f[Reporter] 正在生成最终报告。历史记录: {state[history]}) prompt f 基于以下执行历史和观察结果为用户生成一份完整、结构化的总结报告 初始问题{state[input]} 执行历史 {chr(10).join(state[history])} message [HumanMessage(contentprompt)] response llm.invoke(message) final_report response.content.strip() return {output: final_report, history: [f报告生成完毕: {final_report[:100]}...]} # 2. 构建图 def create_research_agent_graph(): 创建并返回一个研究助理的工作流图。 workflow StateGraph(AgentState) # 添加节点 workflow.add_node(planner, planner_node) workflow.add_node(executor, executor_node) workflow.add_node(reporter, reporter_node) # 设置入口点 workflow.set_entry_point(planner) # 添加边定义流程逻辑 workflow.add_edge(planner, executor) # 从执行器出来需要判断是否还有步骤未完成 workflow.add_conditional_edges( executor, # 条件判断函数如果所有步骤完成前往reporter否则继续执行 lambda state: reporter if len(state.get(completed_steps, [])) len(state.get(plan, [])) else executor, { executor: executor, # 继续循环执行 reporter: reporter # 去生成报告 } ) workflow.add_edge(reporter, END) # 编译图 return workflow.compile() # 快捷方式 research_graph create_research_agent_graph()这个图实现了基本的“规划-执行-报告”循环planner节点先规划步骤。executor节点循环执行每个步骤直到所有步骤标记为完成。reporter节点汇总生成最终报告。4. 实战深化集成工具、安全护栏与可观测性仅有循环不够我们需要Harness工程来加固它。4.1 创建自定义工具Tools工具是智能体与外界交互的“手”。我们创建一个简单的计算器工具并展示如何集成到智能体中。# src/tools/calculator.py from langchain.tools import tool from pydantic import BaseModel, Field # 定义工具的输入模式这能帮助LLM正确生成调用参数 class CalculatorInput(BaseModel): a: float Field(description第一个数字) b: float Field(description第二个数字) operator: str Field(description运算符支持 add, subtract, multiply, divide) tool(args_schemaCalculatorInput) def calculator_tool(a: float, b: float, operator: str) - str: 执行简单的数学计算。 try: if operator add: result a b elif operator subtract: result a - b elif operator multiply: result a * b elif operator divide: if b 0: return 错误除数不能为零。 result a / b else: return f错误不支持的运算符 {operator}。支持: add, subtract, multiply, divide return f计算结果: {a} {operator} {b} {result} except Exception as e: return f计算过程中发生错误: {e}4.2 实现安全护栏Safety Guardrails安全护栏在动作执行前后进行检查。我们实现一个简单的“权限检查”和“输出过滤”护栏。# src/harness/safety_check.py from typing import Dict, Any import re class SafetyHarness: 安全护栏示例。 staticmethod def pre_execution_guard(state: Dict[str, Any], intended_action: str) - tuple[bool, str]: 在执行任何工具或操作前检查。 # 示例1检查是否试图执行危险命令 dangerous_patterns [rrm\s-rf, rformat\sc:, rdrop\sdatabase, rsudo] for pattern in dangerous_patterns: if re.search(pattern, intended_action, re.IGNORECASE): return False, f安全拦截检测到潜在危险操作 {intended_action} # 示例2检查输入是否包含敏感信息简单演示 if password in intended_action or 密钥 in intended_action: return False, 安全拦截输入中可能包含敏感词汇。 print(f[Safety Guard] 预检通过: {intended_action[:50]}...) return True, 预检通过 staticmethod def post_execution_guard(observation: str) - tuple[bool, str]: 在获得工具执行结果后检查。 # 示例过滤结果中的个人身份信息PII - 简单正则示例 # 在实际项目中应使用更专业的PII检测库 pii_patterns { r\b\d{18}\b: [身份证号已屏蔽], r\b1[3-9]\d{9}\b: [手机号已屏蔽], r\b[A-Za-z0-9._%-][A-Za-z0-9.-]\.[A-Z|a-z]{2,}\b: [邮箱已屏蔽], } filtered_obs observation for pattern, replacement in pii_patterns.items(): filtered_obs re.sub(pattern, replacement, filtered_obs) if filtered_obs ! observation: print(f[Safety Guard] 输出内容已进行PII过滤。) return True, filtered_obs # 返回过滤后的结果 return True, observation # 原样返回4.3 集成工具与护栏到工作流节点我们修改之前的executor_node使其能够动态选择并安全地调用工具。# 在 research_agent_graph.py 中更新或新建一个节点 from src.tools.calculator import calculator_tool from src.harness.safety_check import SafetyHarness def enhanced_executor_node(state: AgentState) - dict: 增强的执行节点能调用工具并经过安全护栏检查。 all_steps state[plan] completed state.get(completed_steps, []) remaining [s for s in all_steps if s not in completed] if not remaining: return {observation: 所有步骤已完成。, output: 任务执行完毕。} current_step remaining[0] print(f[Enhanced Executor] 处理步骤: {current_step}) # 1. 预执行安全检查 safe_to_run, pre_check_msg SafetyHarness.pre_execution_guard(state, current_step) if not safe_to_run: return {observation: f执行被安全护栏阻止: {pre_check_msg}, completed_steps: completed [current_step]} # 2. 判断步骤类型并选择行动 observation # 简单启发式如果步骤描述中包含“计算”或数字尝试使用计算器工具 if 计算 in current_step or any(char.isdigit() for char in current_step): # 这里简化处理实际中应该用LLM来解析步骤并生成工具调用参数 # 我们假设步骤是类似“计算一下 123 乘以 456 的结果” try: # 非常简单的提取数字逻辑仅用于演示 import re numbers re.findall(r\d, current_step) if len(numbers) 2: a, b int(numbers[0]), int(numbers[1]) # 调用工具 tool_result calculator_tool.invoke({a: a, b: b, operator: multiply}) observation f使用计算器工具得到: {tool_result} else: observation f无法从步骤{current_step}中提取足够数字进行计算。 except Exception as e: observation f调用计算工具时出错: {e} else: # 其他步骤仍用LLM模拟 prompt f模拟执行研究步骤{current_step} message [HumanMessage(contentprompt)] response llm.invoke(message) observation response.content.strip() # 3. 后执行安全检查与过滤 _, filtered_observation SafetyHarness.post_execution_guard(observation) # 更新状态 new_completed completed [current_step] return { observation: filtered_observation, completed_steps: new_completed, history: [f执行了步骤 {current_step}结果: {filtered_observation[:100]}...] }4.4 添加可观测性日志与监控可观测性让我们能洞察智能体的内部运行。我们实现一个简单的装饰器来记录节点的执行情况。# src/harness/observability.py import time import functools from datetime import datetime class Observability: 可观测性模块用于记录节点执行跟踪。 logs [] staticmethod def log_node_invocation(node_name: str, state_input: dict, duration: float, result: dict): log_entry { timestamp: datetime.utcnow().isoformat(), node: node_name, input_state_snapshot: {k: str(v)[:200] for k, v in state_input.items()}, # 截断长内容 execution_duration_ms: round(duration * 1000, 2), result_state_update: {k: str(v)[:200] for k, v in result.items()}, } Observability.logs.append(log_entry) print(f[Observability] {node_name} 执行耗时: {log_entry[execution_duration_ms]}ms) # 在实际项目中这里可以将日志发送到ELK、Prometheus等系统 staticmethod def track_node(func): 装饰器用于自动跟踪节点函数的执行。 functools.wraps(func) def wrapper(state): start_time time.time() result func(state) end_time time.time() Observability.log_node_invocation(func.__name__, state, end_time - start_time, result) return result return wrapper # 使用装饰器增强节点函数 # 只需在节点函数定义前加上 Observability.track_node # 例如 # Observability.track_node # def planner_node(state: AgentState) - dict: # ...5. 完整运行与测试现在我们将所有部分组合起来创建一个完整的、具备Harness能力的智能体应用。# src/main.py from src.graphs.research_agent_graph import create_research_agent_graph from src.harness.observability import Observability import asyncio async def main(): print( 启动具备Harness的AI研究智能体 ) # 1. 获取编译好的工作流图 # 注意在实际的research_agent_graph.py中我们需要用装饰器包装节点函数 # 这里假设我们已经用 Observability.track_node 装饰了 planner_node, enhanced_executor_node, reporter_node app create_research_agent_graph() # 2. 定义初始状态 initial_state { input: 请研究一下太阳能光伏发电的基本原理并计算如果一块标准光伏板效率20%在日照强度为1000W/m²下工作5小时能产生多少度电假设光伏板面积是2平方米。, plan: [], completed_steps: [], observation: , history: [], output: } # 3. 运行图 print(f\n用户输入: {initial_state[input]}) print(- * 50) final_state None # LangGraph编译的app是可调用对象我们也可以流式获取每个步骤 async for step in app.astream(initial_state, stream_modevalues): node_name list(step.keys())[0] if step else unknown print(f[Graph Step] 节点 {node_name} 执行完毕。) # 最后一个状态就是最终状态 final_state step.get(node_name, {}) print(- * 50) print(\n 任务执行完成 ) print(f最终输出:\n{final_state.get(output, 无输出)}) # 4. 打印可观测性日志 print(\n 执行跟踪日志 ) for i, log in enumerate(Observability.logs): print(f{i1}. [{log[timestamp]}] {log[node]} - {log[execution_duration_ms]}ms) if __name__ __main__: asyncio.run(main())运行这个程序你将看到智能体按照“规划-循环执行-报告”的流程工作并且每一步都有日志输出和安全检查如果触发。这便是一个具备基本循环工程和Harness工程特性的AI智能体。6. 常见问题与排查思路在构建和运行此类智能体系统时你可能会遇到以下典型问题问题现象可能原因排查思路与解决方案图编译失败StateGraph状态模式定义错误节点函数返回值与状态字段不匹配。1. 检查AgentState的TypedDict定义确保字段名拼写正确。2. 确保每个节点函数返回一个字典其键必须是AgentState中定义的字段的子集。智能体陷入无限循环条件边add_conditional_edges的逻辑判断函数有误未能正确终止。1. 在条件函数中打印state进行调试。2. 确保终止条件如len(completed_steps) len(plan)逻辑严密。可以在循环超过N次后强制跳出。工具调用失败或参数错误LLM生成的工具调用参数格式不符合工具期望的args_schema。1. 为工具提供清晰、结构化的描述和参数定义。2. 使用LangChain的StructuredTool或Tool的args_schema特性。3. 在调用工具前可以增加一个“参数验证与修正”节点。状态History未正确累积未使用Annotated[List[str], operator.add]注解来声明可累加字段。确保在状态模式中希望被多个节点追加内容的字段如history使用Annotated和operator.add进行注解。这是LangGraph实现状态聚合的关键。Ollama模型调用超时或无响应Ollama服务未启动模型未正确下载网络问题。1. 运行ollama serve确保服务在运行。2. 运行ollama list确认模型已存在。3. 在代码中尝试使用timeout参数并捕获异常。安全护栏误拦截正常请求正则表达式或规则过于严格。1. 细化规则逻辑区分上下文。2. 实现护栏的“学习模式”或“审核模式”将拦截记录交由人工复审逐步优化规则。执行历史过长导致性能下降history或state随着循环次数增加变得巨大。1. 定期对历史进行摘要Summarize只保留关键信息。2. 将历史存储到外部数据库如Redis状态中只保留引用ID。7. 最佳实践与工程建议将AI智能体投入生产环境需要遵循更严格的软件工程准则。状态设计要精简且聚焦只将真正需要跨节点共享和修改的数据放入State。对于大型数据如检索到的文档全文在状态中存储引用如ID或索引而非数据本身。使用Pydantic进行严格的输入输出验证确保数据类型安全。工具设计遵循单一职责原则每个工具应只做一件事并做好错误处理。工具的描述description要极其精确这直接影响到LLM能否正确调用它。为工具编写单元测试确保其功能稳定。实现分级安全策略基础层语法/规则如本文所示的正则表达式过滤。模型层语义使用一个轻量级的“审查模型”对智能体的计划和输出进行二次评分和过滤。人工层关键操作对于涉及资金、数据删除、对外发布等高风险操作设计“人工审批”节点将操作挂起等待人工确认。建立全面的可观测体系日志记录每个节点的输入/输出、耗时、Token使用量。链路追踪为每次用户会话分配唯一trace_id串联所有相关日志和调用。指标监控监控智能体的平均完成时间、成功率、工具调用分布、错误类型。成本监控密切监控不同模型和工具的调用成本。版本化与回滚将智能体的工作流图Graph、工具集、提示词模板进行版本控制如Git。部署新版本时可以采用蓝绿部署或金丝雀发布策略先让小部分流量走新版本。必须保留快速回滚到上一稳定版本的能力。测试策略单元测试测试每个独立的工具和节点函数。集成测试测试整个工作流图使用固定的输入验证输出是否符合预期。模糊测试/对抗测试输入一些边缘案例、无意义或带有诱导性的问题观察智能体行为是否安全、稳定。通过以上架构和最佳实践你可以构建出不仅“智能”而且“可靠、安全、可维护”的AI智能体系统。这标志着AI应用从原型走向生产的关键一步。 
