在实际 AI 项目开发中我们常常面临一个核心矛盾如何让一个具备强大推理能力的大语言模型LLM稳定、可靠地执行一系列复杂的、多步骤的任务。简单的单次问答无法满足自动化流程、数据分析、系统集成等场景的需求。这时我们就需要引入“智能体Agent”的概念。然而构建一个健壮的智能体并非易事它涉及到任务规划、工具调用、状态管理、错误处理等一系列工程挑战。近年来围绕如何高效构建和管理智能体业界逐渐形成了两种互补的工程范式循环工程Loop Engineering和Harness 工程Harness Engineering。这两种范式并非对立而是从不同层面解决智能体架构的复杂性问题。循环工程聚焦于智能体内部的“思考-行动-观察”循环机制它定义了智能体如何与外部世界工具、API、用户进行交互并基于反馈调整其行为。这关乎智能体执行任务的核心逻辑与状态流转。而 Harness 工程则更侧重于智能体生命周期的“外部”管理包括如何安全地创建、配置、部署、监控和迭代智能体确保其作为一个可交付、可运维的软件组件在复杂环境中稳定运行。理解并实践这两种工程思想是构建企业级、生产可用 AI 智能体的关键。本文将深入探讨循环工程与 Harness 工程的核心概念、技术实现与最佳实践。我们将以一个“企业级安全合规自动化检测”的智能体为例从零开始逐步构建其内部循环逻辑并为其搭建完整的外部管理框架。通过这个过程你将掌握设计一个既能“聪明思考”又能“稳定运行”的 AI 智能体的完整方法论。1. 理解智能体的核心循环工程循环工程是构建智能体执行逻辑的基石。它源于 ReActReasoning Acting等范式核心思想是让智能体在一个循环中不断进行推理、行动并根据行动结果调整下一步计划。1.1 智能体循环的基本模型一个典型的智能体循环包含以下几个关键阶段任务接收与解析智能体接收一个自然语言指令如“检查服务器A的防火墙配置是否符合PCI-DSS标准”并理解其意图和隐含的子任务。规划与推理智能体基于当前任务和已知信息上下文决定下一步该做什么。这可能涉及分解任务、选择工具、生成参数。行动执行智能体调用一个或多个外部工具Tool来执行具体操作例如运行一个SSH命令查询防火墙规则或调用一个REST API获取安全组配置。观察与评估智能体接收工具执行的结果成功、失败、返回数据并评估当前任务完成度。状态更新与循环判断智能体将观察结果整合到内部状态记忆/上下文中并判断任务是否完成。如果未完成则回到第2步继续循环如果完成则输出最终结果。这个循环的健壮性直接决定了智能体的能力上限。一个脆弱的循环可能在工具调用失败时崩溃或在复杂任务中迷失方向。1.2 使用 LangGraph 实现可控循环LangChain 的 LangGraph 是当前实现复杂智能体循环的强力工具。它允许你将智能体的不同组件LLM、工具、状态建模为一个有向图Graph节点代表处理步骤边代表控制流。以下是一个基于 LangGraph 构建的简易安全检测智能体的循环实现框架。我们假设智能体拥有两个工具execute_ssh_command和fetch_compliance_rules。首先定义智能体的状态。状态是一个字典在循环中传递和更新。from typing import TypedDict, Annotated, List from langgraph.graph.message import add_messages import operator class AgentState(TypedDict): # 消息历史用于LLM上下文 messages: Annotated[List, add_messages] # 当前任务描述 task: str # 已收集的检查结果 findings: List[str] # 当前步骤 step: str然后定义图中的各个节点。plan_node负责让 LLM 根据当前状态决定下一步行动。from langchain_core.messages import HumanMessage, AIMessage, ToolMessage from langchain_openai import ChatOpenAI from langchain.tools import tool from langchain_core.prompts import ChatPromptTemplate # 初始化LLM llm ChatOpenAI(modelgpt-4o, temperature0) # 定义工具 tool def execute_ssh_command(host: str, command: str) - str: 在目标主机上执行SSH命令并返回结果。 # 这里应实现真正的SSH客户端逻辑例如使用paramiko # 为示例我们返回模拟数据 if command.startswith(sudo iptables -L): return Chain INPUT (policy DROP)\n target prot opt source destination\n ACCEPT all -- anywhere anywhere state RELATED,ESTABLISHED\n ACCEPT tcp -- 10.0.0.0/8 anywhere tcp dpt:22 return Command executed. tool def fetch_compliance_rules(standard: str) - str: 获取特定合规标准如PCI-DSS的规则列表。 rules_db { PCI-DSS: [1. Install and maintain a firewall., 2. Do not use vendor-supplied defaults.] } return \n.join(rules_db.get(standard, [Standard not found.])) tools [execute_ssh_command, fetch_compliance_rules] llm_with_tools llm.bind_tools(tools) def plan_node(state: AgentState): 规划节点让LLM分析任务决定下一步是调用工具还是结束。 messages state[messages] # 最后一次消息是用户的任务输入 if not messages or not isinstance(messages[-1], HumanMessage): prompt ChatPromptTemplate.from_messages([ (system, 你是一个安全合规检查助手。请分析任务决定下一步是调用工具还是直接回答。可用的工具有execute_ssh_command, fetch_compliance_rules。), (human, {task}) ]) current_message prompt.invoke({task: state[task]}).to_messages() else: # 后续循环中消息历史已包含之前的交互 current_message messages # 请求LLM做出决策 response llm_with_tools.invoke(current_message) # 将LLM的响应添加到消息历史 state[messages].append(response) # 更新步骤状态 if response.tool_calls: state[step] fcalling_tool_{response.tool_calls[0][name]} else: state[step] final_answer return state接下来是execute_tools_node负责执行LLM选择的工具。def execute_tools_node(state: AgentState): 工具执行节点运行LLM请求的工具并将结果封装为ToolMessage。 messages state[messages] last_message messages[-1] tool_messages [] if last_message.tool_calls: for tool_call in last_message.tool_calls: tool_name tool_call[name] tool_args tool_call[args] # 查找并调用对应的工具函数 for t in tools: if t.name tool_name: try: result t.invoke(tool_args) # 将结果作为ToolMessage添加 tool_messages.append(ToolMessage(contentstr(result), tool_call_idtool_call[id])) # 可选将关键发现存入state if tool_name execute_ssh_command: state[findings].append(fSSH检查结果: {result[:100]}...) except Exception as e: tool_messages.append(ToolMessage(contentfTool error: {e}, tool_call_idtool_call[id])) # 将工具执行结果添加到消息历史 state[messages].extend(tool_messages) return state最后我们需要一个should_continue函数来控制循环的走向以及一个final_node来结束循环并输出。from langgraph.graph import END def should_continue(state: AgentState) - str: 路由函数根据最新消息决定下一步是继续执行工具还是结束。 last_message state[messages][-1] # 如果上一步LLM的响应中没有调用工具说明任务完成 if not getattr(last_message, tool_calls, None): return end # 否则继续执行工具 return continue def final_node(state: AgentState): 最终节点整理并输出最终结果。 # 从消息历史中提取最终的LLM回答 final_answer state[messages][-1].content # 也可以整理state[findings]中的信息 report f任务完成报告:\n{final_answer}\n\n收集到的发现:\n \n.join(state[findings]) state[step] completed # 在实际应用中这里可以将报告持久化或返回给调用方 print(report) return state现在我们可以使用 LangGraph 将这些节点组装成一个工作流图。from langgraph.graph import StateGraph, START # 创建图 workflow StateGraph(AgentState) # 添加节点 workflow.add_node(plan, plan_node) workflow.add_node(execute_tools, execute_tools_node) workflow.add_node(final, final_node) # 设置边和路由 workflow.add_edge(START, plan) workflow.add_conditional_edges( plan, should_continue, { continue: execute_tools, end: final } ) workflow.add_edge(execute_tools, plan) # 执行工具后回到规划节点进行下一轮思考 workflow.add_edge(final, END) # 编译图 app workflow.compile()至此一个具备基本“规划-执行-观察”循环的智能体核心就构建完成了。你可以通过以下方式运行它# 初始化状态 initial_state AgentState( messages[], task检查主机192.168.1.100的防火墙配置并对比PCI-DSS标准。, findings[], stepstart ) # 执行智能体 final_state app.invoke(initial_state)这个循环实现了智能体的“大脑”功能但它还非常脆弱缺乏生产级别的鲁棒性管理。这就是 Harness 工程要解决的问题。2. 构建智能体的外壳Harness 工程如果说循环工程赋予了智能体“灵魂”执行逻辑那么 Harness 工程就是为这个灵魂打造一个坚固的“躯壳”或“驾驶舱”Harness。Harness 工程关注的是智能体作为软件服务的非功能性需求安全性、可观测性、配置化、版本管理、资源隔离等。2.1 Harness 的核心职责一个完整的智能体 Harness 通常需要处理以下方面生命周期管理智能体的创建、初始化、启动、暂停、销毁。配置管理集中管理 LLM 模型参数、API密钥、工具列表、提示词模板等。避免将配置硬编码在循环逻辑中。安全与合规输入/输出净化防止提示词注入、过滤敏感信息。工具调用沙箱限制工具可访问的资源如网络、文件系统。权限控制基于角色控制谁能创建、运行、查看哪个智能体。审计日志记录所有的用户交互、工具调用和决策过程以满足合规要求。可观测性监控指标跟踪每次调用的耗时、Token 使用量、工具调用成功率、成本。分布式追踪将一次智能体会话中的所有步骤LLM调用、工具调用关联起来便于调试。结构化日志不仅打印日志还要以结构化的方式记录关键事件和状态变更。持久化与状态管理将会话状态、记忆向量存储到数据库支持长对话和智能体“睡眠/唤醒”。版本控制与回滚对智能体的配置、提示词、工具集进行版本化管理支持快速回滚到稳定版本。资源隔离与多租户在 SaaS 或平台化场景下隔离不同用户或团队的智能体运行环境与数据。2.2 设计一个企业级智能体 Harness我们将基于 Golang 设计一个轻量级但功能完备的 Harness 系统框架用于管理我们之前构建的安全合规检测智能体。选择 Go 是因为其在并发、网络服务和系统编程方面的优势适合构建高并发的控制平面。项目结构 (Monorepo 风格)enterprise-ai-harness/ ├── cmd/ │ ├── harness-server/ # Harness 主服务 │ └── agent-runner/ # 智能体运行时隔离进程 ├── internal/ │ ├── agent/ │ │ ├── manager.go # 智能体生命周期管理 │ │ ├── registry.go # 智能体定义注册中心 │ │ └── state_store.go # 状态存储接口 │ ├── config/ │ │ └── manager.go # 配置管理 │ ├── security/ │ │ ├── sanitizer.go # 输入输出净化 │ │ └── policy_engine.go # 访问控制策略引擎 │ ├── observability/ │ │ ├── tracer.go # 分布式追踪 │ │ └── metrics.go # 指标收集 │ └── storage/ │ └── repository.go # 数据层抽象 ├── pkg/ │ ├── agentproto/ # 智能体相关 Protocol Buffers 定义 │ └── toolplugin/ # 工具插件 SDK ├── deployments/ │ └── docker-compose.yml # 依赖服务DB, Redis, Jaeger └── configs/ └── harness.yaml # Harness 主配置核心组件详解配置管理 (internal/config/manager.go)使用 Viper 等库实现分层配置默认值、文件、环境变量、远程配置中心。关键配置包括# configs/harness.yaml logging: level: info format: json observability: tracing: enabled: true exporter: jaeger endpoint: localhost:6831 metrics: enabled: true port: 9090 agents: security_scanner: runtime: langgraph_python # 指定运行时类型 entry_point: agent_loop.py env: OPENAI_API_KEY: ${OPENAI_API_KEY} resources: max_memory_mb: 512 timeout_seconds: 300 tools: - name: execute_ssh_command allowed_hosts: [10.0.0.0/8] - name: fetch_compliance_rules cache_ttl: 3600安全沙箱与工具执行 (internal/security/)这是 Harness 最关键的模块之一。我们不能让智能体直接调用execute_ssh_command而必须通过一个受控的代理。// internal/security/tool_gateway.go package security import ( context fmt agentproto ) type ToolGateway struct { policyEngine PolicyEngine auditLogger AuditLogger } func (tg *ToolGateway) ExecuteTool(ctx context.Context, req *agentproto.ToolCallRequest) (*agentproto.ToolCallResponse, error) { // 1. 审计日志 tg.auditLogger.LogToolCall(ctx, req.AgentId, req.ToolName, req.Arguments) // 2. 策略检查 if err : tg.policyEngine.CheckToolAccess(ctx, req.AgentId, req.ToolName, req.Arguments); err ! nil { return nil, fmt.Errorf(tool access denied: %w, err) } // 3. 输入净化防止命令注入 sanitizedArgs, err : SanitizeInput(req.ToolName, req.Arguments) if err ! nil { return nil, err } // 4. 根据工具名路由到具体的执行器可能在独立进程中 var result string switch req.ToolName { case execute_ssh_command: // 调用一个受控的、具有严格白名单和命令过滤的SSH客户端服务 result, err executeSSHCommandSafely(sanitizedArgs.Host, sanitizedArgs.Command) case fetch_compliance_rules: // 调用内部合规规则API result, err fetchRulesFromInternalAPI(sanitizedArgs.Standard) default: err fmt.Errorf(unknown tool: %s, req.ToolName) } // 5. 输出过滤脱敏 filteredResult : FilterSensitiveData(result) // 6. 记录结果 tg.auditLogger.LogToolResult(ctx, req.AgentId, req.ToolName, filteredResult, err) return agentproto.ToolCallResponse{ Content: filteredResult, Error: errMsg, }, nil }对于execute_ssh_command这样的高危工具executeSSHCommandSafely函数内部会进行严格的验证主机白名单校验仅允许访问预配置的管理网段。命令黑名单/白名单过滤禁止rm -rf /、dd等危险命令或只允许iptables -L、systemctl status firewalld等只读命令。使用非特权账号执行命令。可观测性集成 (internal/observability/)在每个关键步骤注入追踪和指标。// internal/agent/manager.go 片段 func (m *AgentManager) RunAgentSession(ctx context.Context, sessionId string, userInput string) (*agentproto.SessionOutput, error) { // 开始一个追踪Span ctx, span : tracer.Start(ctx, AgentSession) defer span.End() // 记录指标 startTime : time.Now() defer func() { metrics.AgentSessionDuration.Observe(time.Since(startTime).Seconds()) }() // 获取智能体配置 agentDef, err : m.registry.Get(sessionId) if err ! nil { metrics.AgentSessionErrors.Inc() span.RecordError(err) return nil, err } // 调用具体的智能体运行时如Python进程 output, err : m.runtimeExecutor.Execute(ctx, agentDef, userInput) if err ! nil { metrics.AgentSessionErrors.Inc() } else { metrics.AgentSessionSuccess.Inc() } // 将LLM Token使用量等作为属性记录到Span span.SetAttributes(attribute.Int(token.usage, output.TokenUsage)) return output, err }智能体运行时隔离为了安全性和资源控制智能体的核心循环如我们的Python LangGraph应用应该在独立的、受控的进程中运行。Harness 服务通过 gRPC 或 HTTP 与这些“运行时”通信。// cmd/agent-runner/main.go 概览 func main() { // 1. 加载为该智能体实例分配的配置从Harness服务获取 cfg : loadConfigFromEnv() // 2. 初始化安全的工具执行客户端该客户端只会连接回Harness的ToolGateway toolClient : NewRestrictedToolClient(cfg.GatewayURL) // 3. 加载智能体循环代码如之前的LangGraph app agentApp : langgraph.LoadAgent(cfg.EntryPoint, toolClient) // 4. 启动gRPC服务器等待Harness服务下发任务 server.StartGRPCServer(agentApp) }这种架构实现了“控制平面”Harness与“数据平面”智能体运行时的分离便于升级、扩缩容和安全管控。3. 循环与 Harness 的协同一个完整的工作流现在我们将循环工程与 Harness 工程结合起来看看一次完整的智能体调用是如何进行的。用户请求用户通过 API 向 Harness 服务发送请求“检查服务器 X 的合规性”。Harness 接收与预处理认证/授权验证用户身份和权限。输入净化检查用户输入中是否有恶意内容。创建会话生成唯一的 Session ID初始化审计日志和追踪链路。路由根据请求类型找到对应的“安全合规检测”智能体定义。启动智能体运行时Harness 从池中分配或启动一个agent-runner进程并将智能体配置和用户输入传递过去。智能体循环开始agent-runner中的 LangGraph 应用开始工作。规划LLM 决定首先调用fetch_compliance_rules工具。受控工具调用LangGraph 应用将工具调用请求发送回 Harness 的ToolGateway而非直接执行。ToolGateway执行安全检查、审计、输入净化。安全的执行器从内部 API 获取合规规则结果经过输出过滤后返回给agent-runner。循环继续agent-runner收到工具结果LLM 进行下一步规划决定调用execute_ssh_command。再次受控调用同样经过ToolGateway的安全检查和命令过滤在目标服务器上执行只读的防火墙检查命令。生成最终答案LLM 综合所有信息生成合规性报告。结果返回与后处理agent-runner将最终报告和完整的交互日志返回给 Harness。Harness 进行最终输出净化记录会话完成指标关闭追踪 Span。报告通过 API 返回给用户。所有审计日志、指标和追踪数据被发送到相应的后端系统如 ELK、Prometheus、Jaeger。4. 生产环境部署与运维考量将上述架构投入生产还需要考虑以下关键点4.1 部署架构建议采用容器化部署例如使用 Docker Compose 或 Kubernetes。# deployments/docker-compose.yml 简化版 version: 3.8 services: harness-server: build: ./cmd/harness-server ports: - 8080:8080 environment: - DB_HOSTpostgres - REDIS_HOSTredis - JAEGER_AGENT_HOSTjaeger depends_on: - postgres - redis - jaeger agent-runner-python: build: ./runtimes/python # 非持久化由Harness动态调度 deploy: replicas: 3 environment: - HARNESS_GATEWAY_URLhttp://harness-server:8080 # 严格限制资源与权限 cap_drop: - ALL read_only: true networks: - internal postgres: image: postgres:15 volumes: - pgdata:/var/lib/postgresql/data environment: - POSTGRES_DBharness - POSTGRES_USERharness - POSTGRES_PASSWORDyour_secure_password jaeger: image: jaegertracing/all-in-one:latest ports: - 16686:16686 prometheus: image: prom/prometheus:latest volumes: - ./configs/prometheus.yml:/etc/prometheus/prometheus.yml ports: - 9090:9090 volumes: pgdata:4.2 监控与告警关键指标agent_session_duration_seconds智能体会话耗时分布。agent_session_total,agent_session_errors_total会话总量与错误量。tool_calls_total,tool_call_errors_total工具调用统计。llm_token_usage_totalLLM Token 消耗按模型细分。runner_container_memory_usage运行时容器资源使用情况。告警规则Prometheus Alertmanager工具调用错误率持续5分钟 5%。智能体会话平均延迟超过30秒。LLM API 调用失败。4.3 常见问题排查清单当智能体出现异常时可以遵循以下路径排查问题现象可能原因检查点解决方案智能体不响应或超时1. LLM API 连接失败或限流。2. 工具调用卡死如SSH连接超时。3. 智能体陷入无限循环。1. 查看 Harness 日志中 LLM 调用的错误信息。2. 检查ToolGateway的审计日志看工具调用是否长时间无返回。3. 在追踪系统Jaeger中查看会话的 Span卡在哪一步。1. 检查 LLM API 密钥、网络、配额。2. 为工具调用设置合理的超时时间并实现熔断机制。3. 在智能体循环中设置最大迭代次数。工具调用返回“权限被拒绝”1. 策略引擎拒绝访问。2. 目标主机SSH密钥错误或权限不足。3. 命令不在白名单内。1. 查看ToolGateway的策略检查日志。2. 检查executeSSHCommandSafely函数内的主机白名单和命令过滤器。3. 查看安全执行器自身的错误日志。1. 复核智能体配置和用户权限。2. 确保用于执行命令的服务账号具有必要且最小的权限。3. 更新工具白名单配置。智能体输出不符合预期或“胡言乱语”1. 提示词Prompt设计有误。2. 上下文窗口溢出丢失了关键历史信息。3. 工具返回的数据格式LLM无法解析。1. 检查发送给LLM的最终提示词内容可在追踪日志中查看。2. 检查会话历史是否被正确管理和截断。3. 检查工具返回的结果是否清晰、结构化。1. 迭代优化提示词加入更明确的指令和格式要求。2. 实现智能的上下文窗口管理策略如摘要、选择性记忆。3. 让工具返回结构化的JSON数据并在提示词中指导LLM如何阅读。内存使用持续增长1. 智能体运行时内存泄漏。2. 会话状态未及时清理。3. 大模型加载占用内存。1. 监控agent-runner容器的内存指标。2. 检查state_store的实现是否有缓存未释放。1. 为agent-runner设置严格的内存限制和重启策略。2. 实现会话TTL定期清理过期会话状态。3. 考虑使用更轻量的模型或优化加载方式。4.4 安全与合规最佳实践最小权限原则为每个智能体和工具配置精确的权限。SSH工具只能访问特定网段和命令。审计一切所有用户输入、LLM请求/响应、工具调用/结果、内部决策都必须记录在不可篡改的审计日志中并关联到具体用户和会话。输入/输出过滤对所有来自不可信源用户、工具返回的数据进行严格的验证、转义和过滤防止注入攻击和数据泄露。网络隔离将agent-runner运行时容器部署在独立的内网中严格限制其出站连接只允许与 Harness 服务和必要的内部API通信。配置安全API密钥、密码等敏感信息必须通过 Secrets Manager如 HashiCorp Vault、AWS Secrets Manager动态获取而非写在配置文件中。循环工程与 Harness 工程共同构成了构建可靠 AI 智能体的双翼。循环工程让你专注于让智能体“更聪明”通过精巧的流程设计解决复杂问题而 Harness 工程则确保这个聪明的智能体在真实、复杂且充满风险的环境中能够“安全、稳定、可控”地运行。从简单的脚本到企业级系统正是对 Harness 工程的深入理解和实践拉开了原型与产品之间的差距。在开始你的下一个 AI 智能体项目时不妨从设计它的 Harness 开始思考这会让你的架构之路走得更稳、更远。