1. 先搞清楚“循环工程”和“Harness工程”到底在解决什么问题如果你最近在关注AI智能体开发大概率会看到“循环工程”和“Harness工程”这两个词。它们听起来很学术但背后指向的是同一个核心痛点如何让一个AI智能体Agent不只是“跑通一次”而是能稳定、可靠、可管理地处理真实世界里的复杂任务。“循环工程”强调的是智能体在执行任务时的闭环反馈与迭代能力。一个只会按预设脚本走一遍的Agent遇到意外情况就卡住了。而具备循环能力的Agent能根据执行结果比如代码运行报错、网页操作失败、API返回异常进行判断、调整策略并再次尝试直到任务完成或达到终止条件。这就像一个有经验的程序员写代码、运行、看报错、改bug、再运行形成一个循环。“Harness工程”则更侧重于为智能体提供一套标准化的“装备”或“框架”让它能安全、合规、高效地接入各种工具和环境。你可以把Harness想象成一套标准化的宇航服或潜水装备它定义了智能体如何与外部世界操作系统、浏览器、数据库、API安全交互的接口、协议和边界。没有这套“装备”智能体直接裸奔操作生产环境风险极高。所以当这两个概念放在一起时它们共同构成了企业级AI智能体架构的基石一个负责内在的决策与自适应循环一个负责外在的安全与标准化交互。这篇文章我就以一个实际构建过智能体系统的角度拆解一下如何将这两者落地而不是停留在概念讨论。2. 从零搭建环境准备与核心组件选型在动手之前先别急着写代码。AI智能体项目对环境的整洁度和依赖管理要求很高混乱的环境是后期一切“玄学”问题的根源。我建议采用Monorepo单体仓库来管理这能极大简化依赖和模块间的引用。2.1 基础环境与项目结构首先确保你的开发机满足基本条件。AI智能体项目通常对内存和网络要求更高因为需要同时运行大模型服务、向量数据库和各种工具。# 1. 系统与环境 # 推荐使用 Linux (Ubuntu 20.04) 或 macOSWindows建议使用WSL2。 # 内存建议16GB以上因为除了模型还要跑数据库和多个服务。 # 2. 创建并初始化Monorepo项目 mkdir ai-agent-platform cd ai-agent-platform pnpm init # 或 npm/yarn 推荐pnpm用于Monorepo项目结构可以这样规划ai-agent-platform/ ├── packages/ │ ├── core-agent/ # 智能体核心逻辑循环引擎 │ ├── tool-harness/ # 工具套件Harness定义与实现 │ ├── memory-store/ # 向量数据库与记忆管理 │ ├── api-gateway/ # 对外API接口 │ └── shared/ # 共享类型、工具函数 ├── apps/ │ └── demo-web/ # 演示前端可选 ├── docker-compose.yml # 开发环境服务Postgres, Redis等 ├── package.json └── README.md使用pnpm-workspace.yaml或npm workspaces来配置工作区这是管理多包依赖的关键。2.2 核心依赖选择模型、框架与工具链这是选型的关键直接决定开发体验和系统能力上限。大模型服务Agent的“大脑”本地部署Ollama是当前最方便的选择。它帮你管理模型文件提供类OpenAI的API接口。对于开发测试Llama 3.1、Qwen 2.5等7B/8B参数模型足够。云端APIOpenAI GPT-4o、Anthropic Claude、国内各大厂平台。适合生产环境但需考虑成本、延迟和合规。关键点不要只依赖一个模型。在架构设计时抽象出统一的LLM Provider接口方便切换和降级。智能体框架“循环工程”的脚手架LangGraph来自LangChain专为构建有状态、多步骤的智能体而设计。它用“图”Graph来定义工作流节点是步骤边是条件判断天然支持循环。这是实现“循环工程”的首选工具。LlamaIndex更擅长于数据连接和检索RAG可以与之结合。自定义框架如果你用Go或Rust可能会选择更轻量的库但核心思想一致——管理状态和流程。工具与安全层“Harness工程”的体现这是最需要精心设计的部分。每个工具如执行Shell命令、读写文件、调用API、操作浏览器都必须被“Harness”包装。核心原则工具函数不能直接获得原始环境权限。必须通过一个安全沙箱或权限代理层来执行。例如一个“执行命令行”的工具输入应该是经过校验的命令字符串白名单而不是任意字符串。执行环境应该是受限的容器或子进程并有超时、输出大小限制。记忆与状态存储短期记忆/对话历史Redis 或内存存储速度快。长期记忆/向量检索PostgreSQLpgvector扩展或专业的Qdrant、Weaviate、Chroma。用于存储和检索智能体过去的经验、知识片段。编排与部署开发期用docker-compose一键拉起所有服务Ollama, Postgres, Redis。生产环境考虑Kubernetes或成熟的云服务确保智能体服务的可用性和可扩展性。3. 实战拆解一用LangGraph构建“循环”智能体现在我们进入“循环工程”的实战。假设我们要构建一个能自动修复简单Python代码错误的智能体。3.1 定义智能体的状态与工具首先定义智能体在整个任务循环中需要记住什么。这被称为“状态”State。# 在 core-agent/src/state.py 中 from typing import TypedDict, List, Annotated from langgraph.graph.message import add_messages import operator class AgentState(TypedDict): # 任务描述 task: str # 消息历史用于记录与模型的对话 messages: Annotated[List, add_messages] # 当前要分析的代码 current_code: str # 从工具执行中获得的输出如错误信息、测试结果 tool_output: str # 任务是否完成 finished: bool然后定义工具。这里我们先定义一个安全的代码执行工具Harness的雏形。# 在 tool-harness/src/code_runner.py 中 import subprocess import tempfile import os from pathlib import Path class SafeCodeRunner: 一个受限的代码执行工具Harness staticmethod def run_python_code(code: str, timeout: int 10) - dict: 在临时目录中安全地执行一段Python代码。 返回包含 stdout, stderr, returncode 的字典。 result {stdout: , stderr: , returncode: -1} with tempfile.TemporaryDirectory() as tmpdir: code_file Path(tmpdir) / script.py code_file.write_text(code) try: # 使用子进程限制资源 completed subprocess.run( [python, str(code_file)], capture_outputTrue, textTrue, timeouttimeout, cwdtmpdir, # 可以在这里设置更多的安全限制如env, preexec_fn ) result.update({ stdout: completed.stdout, stderr: completed.stderr, returncode: completed.returncode }) except subprocess.TimeoutExpired: result[stderr] Execution timeout exceeded. except Exception as e: result[stderr] fExecution failed: {str(e)} return result3.2 构建LangGraph工作流循环的核心接下来用LangGraph将状态和工具连接成一个可循环的工作流。# 在 core-agent/src/graph.py 中 from langgraph.graph import StateGraph, END from .state import AgentState from tool_harness.src.code_runner import SafeCodeRunner class CodeFixAgentGraph: def __init__(self, llm): self.llm llm self.runner SafeCodeRunner() self.graph self._build_graph() def _call_llm(self, state: AgentState): 节点函数调用大模型分析问题或生成修复代码 # 1. 构建给模型的提示词 messages state[messages] # 添加上下文任务、当前代码、上次工具输出 context_msg { role: user, content: f 任务{state[task]} 当前代码 python {state[current_code]} 上次执行结果 {state[tool_output]} 请分析错误原因并给出修正后的完整代码。如果认为代码已正确请说明。 } messages.append(context_msg) # 2. 调用LLM response self.llm.invoke(messages) # 3. 更新状态 new_messages messages [{role: assistant, content: response.content}] # 4. 尝试从模型回复中提取代码块这里简化处理 new_code self._extract_code_from_response(response.content) or state[current_code] return { messages: new_messages, current_code: new_code, tool_output: # 清空等待下次执行 } def _run_code(self, state: AgentState): 节点函数执行当前代码 result self.runner.run_python_code(state[current_code]) output fSTDOUT:\n{result[stdout]}\nSTDERR:\n{result[stderr]}\nReturn Code: {result[returncode]} # 判断是否成功简单以returncode为0判断 is_success (result[returncode] 0 and not result[stderr]) return { tool_output: output, finished: is_success # 如果成功则标记完成 } def _decide_next(self, state: AgentState): 边函数根据执行结果决定下一步 if state[finished]: return end else: # 如果没完成且错误输出不是超时或致命错误就继续循环分析 if Execution timeout in state[tool_output] or Execution failed in state[tool_output]: return end # 无法处理的错误终止 return analyze # 继续分析 def _build_graph(self): workflow StateGraph(AgentState) # 添加节点 workflow.add_node(analyze, self._call_llm) workflow.add_node(execute, self._run_code) # 设置入口点 workflow.set_entry_point(analyze) # 添加边定义流程 workflow.add_edge(analyze, execute) # 关键条件边实现循环 workflow.add_conditional_edges( execute, self._decide_next, { end: END, analyze: analyze } ) return workflow.compile() def run(self, task: str, initial_code: str): 运行智能体 initial_state: AgentState { task: task, messages: [], current_code: initial_code, tool_output: , finished: False } # 运行图 for event in self.graph.stream(initial_state): # 这里可以打印或记录每一步的状态 print(fStep: {event}) final_state self.graph.invoke(initial_state) return final_state这个图定义了一个清晰的循环分析调用LLM - 执行运行代码- 判断 - 回到分析或结束。这就是“循环工程”的代码体现。智能体会不断尝试直到代码运行成功或遇到无法处理的错误。4. 实战拆解二设计安全的工具“Harness”上面的SafeCodeRunner只是一个简单的例子。真正的“Harness工程”需要更系统的设计。目标是任何工具调用都必须经过授权、验证、隔离和审计。4.1 工具Harness的通用接口首先定义一个所有工具都必须实现的基类接口# 在 tool-harness/src/base.py 中 from abc import ABC, abstractmethod from typing import Any, Dict from pydantic import BaseModel, Field class ToolHarness(ABC): 工具Harness基类 name: str description: str # 使用Pydantic模型定义严格的输入参数模式 args_schema: type[BaseModel] def __init__(self, **kwargs): # 初始化可能需要的配置如API密钥、权限令牌、资源限制 self.config kwargs abstractmethod async def execute(self, validated_args: Dict[str, Any]) - Dict[str, Any]: 执行工具的核心方法。 输入是经过校验的参数。 返回一个标准化的结果字典。 pass def validate_and_run(self, raw_args: Dict[str, Any]) - Dict[str, Any]: 对外暴露的安全入口先校验后执行 # 1. 参数校验 try: validated self.args_schema(**raw_args) except Exception as e: return {success: False, error: f参数校验失败: {e}, data: None} # 2. 权限检查可根据工具类型和用户上下文实现 if not self._check_permission(validated): return {success: False, error: 权限不足, data: None} # 3. 执行可加入超时、资源限制包装 try: # 例如使用asyncio.wait_for实现超时 result await self.execute(validated.dict()) return {success: True, error: None, data: result} except Exception as e: # 记录详细日志但返回给用户的信息要经过过滤 self._log_error(e) return {success: False, error: 工具执行内部错误, data: None} def _check_permission(self, args): # 实现具体的权限逻辑例如检查用户角色、操作范围等 # 这是一个简化示例 return True def _log_error(self, error): # 结构化日志记录用于审计和排查 print(f[Tool Error] {self.name}: {error})4.2 实现几个具体的Harness工具1. 文件读写Harness# tool-harness/src/file_tool.py import os from pathlib import Path from pydantic import BaseModel, Field from .base import ToolHarness class FileReadArgs(BaseModel): filepath: str Field(..., description要读取的文件路径相对或绝对路径) max_size_mb: int Field(5, description允许读取的最大文件大小MB) class FileReadHarness(ToolHarness): name read_file description 读取指定文本文件的内容 args_schema FileReadArgs async def execute(self, validated_args): path Path(validated_args.filepath) max_bytes validated_args.max_size_mb * 1024 * 1024 # 安全检查1路径规范化防止目录遍历攻击 try: resolved_path path.resolve() # 可以设置一个允许访问的根目录白名单 allowed_root Path(/safe/directory) if not str(resolved_path).startswith(str(allowed_root)): raise PermissionError(访问路径超出允许范围) except Exception as e: raise ValueError(f路径安全检查失败: {e}) # 安全检查2文件大小限制 if resolved_path.stat().st_size max_bytes: raise ValueError(f文件大小超过限制 {validated_args.max_size_mb}MB) # 执行读取 content resolved_path.read_text(encodingutf-8, errorsignore) return {content: content, file_size: resolved_path.stat().st_size}2. 网络请求Harness# tool-harness/src/http_tool.py import aiohttp import asyncio from pydantic import BaseModel, Field, HttpUrl from .base import ToolHarness class HttpGetArgs(BaseModel): url: HttpUrl Field(..., description请求的URL) timeout: int Field(30, description请求超时时间秒) allowed_domains: list[str] Field(default_factorylist, description允许访问的域名白名单) class HttpGetHarness(ToolHarness): name http_get description 发起HTTP GET请求 args_schema HttpGetArgs async def execute(self, validated_args): url str(validated_args.url) # 域名白名单检查 if validated_args.allowed_domains: from urllib.parse import urlparse domain urlparse(url).netloc if domain not in validated_args.allowed_domains: raise ValueError(f域名 {domain} 不在白名单内) async with aiohttp.ClientSession() as session: try: async with session.get(url, timeoutvalidated_args.timeout) as response: text await response.text() return { status: response.status, headers: dict(response.headers), body: text[:5000] # 限制返回体大小 } except asyncio.TimeoutError: raise TimeoutError(f请求超时 ({validated_args.timeout}s)) except Exception as e: raise ConnectionError(f网络请求失败: {e})3. 工具注册与管理中心# tool-harness/src/registry.py class ToolRegistry: 工具注册中心统一管理所有Harness def __init__(self): self._tools: Dict[str, ToolHarness] {} def register(self, tool: ToolHarness): if tool.name in self._tools: raise ValueError(f工具名 {tool.name} 已存在) self._tools[tool.name] tool def get_tool(self, name: str) - ToolHarness: tool self._tools.get(name) if not tool: raise KeyError(f未找到工具: {name}) return tool def list_tools(self): return [{name: t.name, desc: t.description} for t in self._tools.values()] # 初始化注册中心 registry ToolRegistry() registry.register(FileReadHarness()) registry.register(HttpGetHarness()) # ... 注册其他工具4.3 将Harness集成到智能体现在我们需要让LangGraph智能体能够安全地调用这些被Harness包装的工具。# 在 core-agent/src/agent_with_harness.py 中 from langchain.tools import StructuredTool from tool_harness.src.registry import registry def create_harness_tool_for_agent(tool_name: str): 将一个Harness工具包装成LangChain可识别的Tool对象 harness registry.get_tool(tool_name) def _wrapper(**kwargs): # 同步调用异步方法实际生产环境建议用异步框架 import asyncio # 注意这里简化了异步调用实际应整合到异步工作流中 loop asyncio.new_event_loop() asyncio.set_event_loop(loop) try: result loop.run_until_complete(harness.validate_and_run(kwargs)) if result[success]: return str(result[data]) # LangChain Tool期望返回字符串 else: return fTool Error: {result[error]} finally: loop.close() # 使用Pydantic模型来自动生成工具的描述和参数schema供LLM理解 return StructuredTool.from_function( func_wrapper, nameharness.name, descriptionharness.description, args_schemaharness.args_schema, ) # 在构建智能体时将工具注入 from langchain.agents import create_react_agent, AgentExecutor from langchain import hub def build_agent(llm): # 从工具注册中心动态创建工具列表 tools [] for tool_info in registry.list_tools(): tools.append(create_harness_tool_for_agent(tool_info[name])) # 使用ReAct代理模式 prompt hub.pull(hwchase17/react) agent create_react_agent(llm, tools, prompt) agent_executor AgentExecutor(agentagent, toolstools, verboseTrue, handle_parsing_errorsTrue) return agent_executor这样智能体在决定调用read_file工具时实际上调用的是经过层层校验和安全包装的FileReadHarness而不是直接操作文件系统。这就是“Harness工程”的价值——将危险的能力关进制度的笼子。5. 生产级考量监控、评估与部署一个能“循环”且“安全”的智能体原型跑起来只是第一步。要用于生产必须解决稳定性、可观测性和规模化问题。5.1 监控与可观测性智能体是个黑盒不行。必须让它透明。结构化日志记录每个工具调用的输入、输出、耗时、状态。使用JSON格式方便接入ELK或Loki。# 在ToolHarness基类的validate_and_run中增强日志 def validate_and_run(self, raw_args): log_entry { timestamp: datetime.utcnow().isoformat(), tool: self.name, args: str(raw_args), # 注意脱敏 user_id: self.context.get(user_id), trace_id: self.context.get(trace_id) } # ... 执行逻辑 log_entry.update({success: result[success], duration_ms: duration}) self.logger.info(json.dumps(log_entry))链路追踪Tracing为每个用户会话或任务分配唯一的trace_id贯穿智能体的所有步骤LLM调用、工具执行、循环迭代。使用OpenTelemetry等标准将追踪数据发送到Jaeger或Zipkin。关键指标Metrics任务成功率任务完成数 / 任务总数。平均循环次数完成一个任务平均需要几次“分析-执行”循环。工具调用延迟每个工具的平均耗时。Token消耗每次LLM调用的输入/输出Token数是成本核心。错误分类权限错误、工具错误、模型错误、超时错误等。5.2 智能体的评估与测试如何知道你的智能体变好了还是变差了单元测试针对工具Harness测试每个工具在正常、边界、异常输入下的行为。集成测试针对工作流用一组固定的“黄金标准”任务Golden Dataset来测试整个智能体。记录每次迭代的最终结果、步骤数和成本。任何代码更新后都跑一遍这个测试集监控指标变化。模糊测试与对抗测试故意输入一些奇怪、恶意或模糊的指令观察智能体是否会做出危险操作或陷入死循环。这是检验Harness安全性的重要手段。人工评估A/B测试对于关键任务将新旧版本智能体的输出结果给真人评估判断哪个更好。5.3 部署与架构模式对于稍复杂的系统单体应用很快会遇到瓶颈。考虑更清晰的架构分离智能体编排服务Agent Orchestrator一个独立的服务负责接收任务实例化LangGraph工作流管理状态机。使用Celery、Dramatiq或直接使用异步框架如FastAPI Background Tasks处理任务队列。将智能体状态持久化到Redis或数据库中支持长时间运行的任务和断点续跑。工具网关Tool Gateway所有工具调用不直接发生而是通过一个统一的网关服务。网关负责最终的身份认证、权限校验、限流、熔断和审计。即使智能体服务被攻破网关是最后一道防线。工具网关可以按权限等级对同一工具提供不同能力的接口如开发环境工具权限高生产环境权限极低。模型网关Model Gateway抽象LLM调用实现多模型路由、负载均衡、失败重试、缓存和降级如GPT-4超时则降级到Claude 3.5 Sonnet。记忆服务Memory Service将向量检索、对话历史、长期记忆抽取为独立服务供多个智能体实例共享。一个简化的生产架构可能如下所示[用户请求] - [API Gateway] - [智能体编排服务] | v [模型网关] - [LLM Service(s)] | v [工具网关] - [Safe Tool Executors] | v [记忆服务] - [Vector DB / Redis]6. 避坑指南与经验之谈最后分享几个从零搭建这类系统时最容易踩坑的地方。1. 状态管理是魔鬼LangGraph的状态State设计决定了智能体的“记忆力”。一开始就要想清楚哪些信息需要跨步骤保留哪些是临时变量状态太大会影响性能太小则智能体会“失忆”。建议将状态分为“会话状态”当前任务相关和“长期记忆”存入向量库定期清理会话状态。2. 工具权限要“最小化”给智能体开文件读写权限时最容易出事。绝对不要给根目录或家目录的写权限。建议为每个项目或会话创建独立的临时沙箱目录任务结束后自动清理。网络工具必须设置严格的域名白名单和超时时间。3. 循环必须有“终止条件”一个设计不好的循环会让智能体在错误里无限尝试烧光你的API预算。必须在循环判断节点_decide_next设置硬性终止条件最大迭代次数如10次、超时总时长、特定错误类型如“语法错误已修复但逻辑错误持续存在”。4. LLM的“幻觉”会传染给工具调用模型可能会生成一个根本不存在的工具名或编造工具的参数格式。建议在将模型输出解析为工具调用前做一次严格的校验。只从已注册的工具列表中选择并用Pydantic强制校验参数。解析失败时让模型重试而不是直接崩溃。5. 不要忽视“人机回环”全自动智能体在复杂场景下必然出错。关键设计在循环中插入“人工审核节点”。当智能体信心不足、或触发了高风险操作如删除文件、调用付费API时自动暂停将决策权交给人类等待批准后再继续。这比事后补救成本低得多。6. 版本化一切智能体工作流LangGraph图、工具Harness、提示词模板都应该进行版本控制。每次改动都可能影响最终输出。建议建立简单的版本管理至少能快速回滚到上一个稳定版本。构建一个健壮的AI智能体系统“循环工程”保证了其智能和韧性“Harness工程”保证了其安全和可控。这两者不是选择题而是必须同时做好的必答题。从一个小而美的原型开始先让单任务循环跑通再逐步加固工具的安全边界最后才考虑性能和规模扩展。这样步步为营才能得到一个真正可用、敢用的AI智能体。