大模型跑通不难难的是把它做成一个稳定、可控、能接业务的企业级服务。这次我们来看 Harness 架构它不是什么新框架而是一套把“模型调用、工具执行、上下文管理、失败重试、日志追踪”统一收口的工程化模式。结合最近社区里频繁提到的 DeepAgent、Agent 开发路线这篇文章会从架构概念讲到企业级项目落地再给出一套可以在本机验证的部署、测试和排错流程。如果只关心单轮问答确实用不到 Harness但一旦涉及多工具编排、批量任务、接口 API 和长期运行Harness 的价值会非常明显。本文不打算讲营销层面的“吊打付费”而是把 Harness 在 AI 大模型和 Agent 开发里到底怎么理解、怎么落地讲清楚。文章会先给出核心能力速览然后用一套可以照做的项目结构演示如何搭建 Agent 服务最后覆盖接口调用、批量任务、资源占用观察和常见问题排查。适合准备做 AI Agent 应用开发、想从 Demo 走向工程化的读者。1. Harness 架构核心能力速览能力项说明架构定位Agent 执行控制层工程化模式不是某个单一开源项目核心功能提示词模板管理、工具注册与调度、上下文管理、执行循环、失败重试、日志追踪典型应用企业知识库问答、自动化报表、智能客服、批量文档处理、多步骤任务编排运行环境CPU 可运行推理加速推荐 GPU具体显存按模型版本测试启动方式Python 命令行启动 / FastAPI 服务启动 / Celery 批量任务节点接口能力通过 HTTP API 对外暴露 Agent 调用能力批量任务支持任务队列目录扫描批量处理或消息队列分发是否支持一键启动依赖项目封装方式建议使用脚本加虚拟环境适合人群后端开发者、AI 应用工程师、算法工程师、想要规范 Agent 项目的团队合规边界自动操作外部系统前必须授权生成内容需人工复核数据隐私按企业规范执行从材料来看Harness 在社区里常与 DeepAgent、Agent 框架放在一起讨论本质上是解决“模型输出不可控、工具调用不稳定、长任务容易断”的问题。架构本身不绑定特定大模型适合作为 AI 大模型应用开发的通用底座。2. 适用场景与使用边界Harness 架构最适合的是有一定流程规律、需要模型反复调用工具的任务。比如一个客服机器人需要先查订单、再查物流、最后生成回复或者一个数据分析 Agent需要按用户问题选择 SQL 查询、Python 计算、图表生成。这类场景中模型负责判断下一步Harness 负责保证每一步都能执行、可重试、可追踪。它在这些场景里效果明显多步骤任务需要多次模型推理和工具调用而不是一次问答。批量处理大量输入需要走同一套 Agent 流程。接口化交付业务方需要把 Agent 能力封装成 HTTP API。可控性要求高生成结果需要记录完整调用链方便回查。不适合的场景同样明确。单次短问答没必要引入 Harness直接用模型 API 更省事需要毫秒级响应的在线接口也不适合把 Agent 编排放在同步链路里另外如果业务本身没有明确工具边界Agent 自由调用工具反而容易出问题。安全边界必须强调。Agent 一旦具备工具调用能力就意味着它能操作文件、访问数据库、调用第三方接口。企业内部使用时需要对工具做白名单控制给每个工具设置权限等级涉及用户隐私数据时只允许在合规环境内处理自动发送消息、自动下单、自动发布内容等操作必须经过人工审批环节。所有生成内容在正式上线前都应该做抽样复核。3. Harness 架构到底在解决什么问题把模型直接接到业务里通常会遇到几个问题。第一个是提示词散落。每个业务场景都要写一段很长的系统提示词散落在代码里改一次要动代码。Harness 把提示词模板独立成管理模块按场景、按版本维护。第二个是工具调用不稳定。模型输出工具调用参数时经常出现格式错误、参数缺失、调用了不存在的工具。Harness 在模型和工具之间加一层校验先解析模型输出再匹配工具定义最后执行工具并记录结果。第三个是上下文管理困难。多轮对话中历史消息、工具返回结果、临时变量混在一起容易把上下文撑爆。Harness 负责把工具结果整理成结构化内容决定哪些进上下文、哪些截断、哪些摘要。第四个是失败重试缺失。工具请求第三方超时是直接返回错误还是重试Harness 统一处理超时、限流、重试避免业务代码到处写 try-catch。第五个是无法观测。模型在做什么、调了哪些工具、花了多长时间、消耗了多少 token如果没有统一日志排查问题非常痛苦。Harness 在每次执行循环里记录完整轨迹。从架构层次看一个典型的 Agent Harness 包含这么几层层级职责关键模块交互层接收用户输入返回最终结果WebUI、HTTP API、命令行执行控制层维护 Agent 主循环决定下一步动作Harness 核心、状态机、循环调度模型接入层对接不同大模型统一输入输出OpenAI 兼容接口、本地模型服务工具层注册和执行业务能力搜索、数据库、计算、文档解析等记忆层管理短期和长期上下文会话缓存、向量库、摘要器可观测层日志、追踪、评估调用链日志、指标采集、错误报告DeepAgent 在社区语境里通常指具备深度任务规划能力的 Agent核心是让执行控制层支持更复杂的任务拆解、长周期执行、反思和校验。它的工程实现依然依赖 Harness 这套底座只是在规划算法和记忆机制上做得更深。4. 企业级 Agent 项目目录设计与模块拆分一个可以在本机跑起来、又能逐步演进成企业级服务的 Agent 项目目录结构建议这样拆分。这里以 Python 为例实际项目可以按团队习惯调整。agent-project/ ├── app.py # FastAPI 入口 ├── config.py # 全局配置 ├── requirements.txt # 依赖清单 ├── prompts/ # 提示词模板目录 │ ├── system_planner.txt │ ├── system_tool_call.txt │ └── summary.txt ├── tools/ # 工具定义 │ ├── __init__.py │ ├── registry.py # 工具注册中心 │ ├── web_search.py # 搜索工具 │ ├── sql_runner.py # 数据库工具 │ └── file_reader.py # 文件读取工具 ├── harness/ │ ├── __init__.py │ ├── loop.py # Agent 主循环 │ ├── context.py # 上下文管理 │ ├── parser.py # 模型输出解析 │ └── retry.py # 重试策略 ├── memory/ │ ├── __init__.py │ └── buffer.py # 会话缓冲 ├── api/ │ ├── __init__.py │ └── routes.py # HTTP 路由 ├── jobs/ # 批量任务 │ ├── __init__.py │ ├── consumer.py # 任务消费者 │ └── scanner.py # 目录扫描 ├── logs/ # 运行日志 ├── inputs/ # 批量输入素材 └── outputs/ # 批量输出结果模块拆分的核心原则提示词、工具、执行逻辑、接口、批量任务互不耦合。业务方新增一个能力时只需要在tools/下新增一个工具并注册调整模型行为时只改prompts/下的模板文件要接新前端不需要改动 Harness 核心。工具注册中心是其中一个关键模块。每个工具都声明名称、描述、参数 URL、权限级别模型根据这些描述决定是否调用。# tools/registry.py from dataclasses import dataclass, field from typing import Callable, Any dataclass class Tool: name: str description: str parameters: dict permission: str read handler: Callable[..., Any] None class ToolRegistry: def __init__(self): self._tools {} def register(self, tool: Tool): self._tools[tool.name] tool def get(self, name: str) - Tool: return self._tools.get(name) def list_tools(self): return [ { name: t.name, description: t.description, parameters: t.parameters, } for t in self._tools.values() ] registry ToolRegistry()这个示例展示了工具注册的基本思路。实际项目中可以在registry.register装饰器里完成工具声明每增加一个业务能力就注册一次Harness 主循环从注册中心获取工具列表并生成给模型的工具描述。5. Harness 本地部署环境准备Harness 架构本身不依赖特定硬件核心依赖是大模型推理服务。如果使用云端模型 API常规开发机即可如果全部本地部署需要按模型规模准备 GPU。更稳妥的判断是先跑通一个小尺寸模型再评估是否升级硬件。环境准备清单如下检查项建议操作系统Linux 服务器优先Windows/macOS 可用于本地开发Python 版本3.10 或 3.11依赖管理pip virtualenv 或 conda大模型服务OpenAI 兼容 API 或本地 Ollama / vLLMCUDA使用本地 GPU 时安装对应驱动和 CUDA磁盘空间依赖包 2GB 左右本地模型按实际大小预留端口规划API 服务默认建议 8000避免冲突Python 虚拟环境创建示例python3 -m venv venv source venv/bin/activate pip install --upgrade pip pip install fastapi uvicorn pydantic requests openai python-dotenv如果使用本地模型需要提前启动模型推理服务。Ollama 启动方式非常直接ollama pull qwen2.5:7b ollama serve模型服务启动后用curl http://127.0.0.1:11434/api/tags验证是否就绪。如果返回模型列表说明推理服务可访问。显存占用以实际模型版本和推理参数为准先跑一次短问题观察nvidia-smi的占用再决定是否增大批量。6. 安装部署与启动方式先准备环境变量文件配置模型接入地址和 API Key。这里以 OpenAI 兼容接口为例# .env 示例 MODEL_NAMEqwen2.5:7b OPENAI_API_BASEhttp://127.0.0.1:11434/v1 OPENAI_API_KEYollama APP_HOST127.0.0.1 APP_PORT8000如果使用云端模型把OPENAI_API_BASE换成云厂商兼容地址OPENAI_API_KEY换成真实密钥。注意不要把密钥提交到代码仓库。接着启动 Agent API 服务。以 FastAPI 为例入口如下# app.py import os from fastapi import FastAPI from pydantic import BaseModel app FastAPI(titleAgent Harness Service) class AgentRequest(BaseModel): session_id: str default user_input: str stream: bool False class AgentResponse(BaseModel): session_id: str output: str tool_calls: list trace_id: str app.post(/agent/run, response_modelAgentResponse) async def agent_run(req: AgentRequest): # 实际项目在这里调用 harness.loop.run() # 返回结果应包含最终回复、工具调用记录和追踪 ID return AgentResponse( session_idreq.session_id, outputAgent 执行结果, tool_calls[], trace_idtrace-001 ) if __name__ __main__: import uvicorn uvicorn.run(app, hostos.getenv(APP_HOST, 127.0.0.1), portint(os.getenv(APP_PORT, 8000)))启动命令source venv/bin/activate python app.py启动后访问http://127.0.0.1:8000/docs如果能看到 Swagger 文档说明服务启动成功。第一次启动建议先关闭其他占用 8000 端口的进程避免端口冲突。如果需要在后台运行服务并保留日志nohup python app.py logs/app.log 21 echo $! logs/app.pid停止服务时先读取 PID 再杀掉进程避免出现进程残留。7. 功能测试与效果验证服务启动后按照“先单轮、再多轮、再工具调用、再批量”的顺序进行验证。每个环节都记录输入、输出、耗时和显存占用作为后续调优的依据。7.1 单轮问答测试第一步先验证模型连通性和基础回答质量。import requests url http://127.0.0.1:8000/agent/run payload { session_id: test-basic, user_input: 用一句话解释什么是 Harness 架构 } response requests.post(url, jsonpayload, timeout120) print(response.json())预期结果是返回一段通顺的中文解释。如果模型服务未配置好这里通常会报连接错误或超时优先检查OPENAI_API_BASE是否可达。7.2 多轮对话测试多轮测试主要验证会话上下文是否被正确保存。payload_1 { session_id: test-multi, user_input: 我要做企业知识库问答系统先帮我列出三个核心模块 } response_1 requests.post(url, jsonpayload_1, timeout120) print(response_1.json()[output]) payload_2 { session_id: test-multi, user_input: 第一个模块的具体实现步骤是什么 } response_2 requests.post(url, jsonpayload_2, timeout120) print(response_2.json()[output])预期结果是第二轮回答能关联第一轮提到的三个模块。如果第二轮回答跑题检查上下文管理模块是否把历史消息传给了模型。7.3 工具调用测试工具调用是 Harness 架构的重头戏。在工具注册中心添加一个模拟查询工具然后让 Agent 调用它。def mock_order_query(order_id: str): return {order_id: order_id, status: 已发货, logistics: 顺丰速运} registry.register(Tool( namequery_order, description根据订单号查询订单状态, parameters{order_id: {type: string, description: 订单号}}, permissionread, handlermock_order_query ))测试输入{ session_id: test-tool, user_input: 帮我查一下订单 20240001 现在什么状态 }预期结果是 Agent 先解析出需要调用query_order工具然后返回订单状态。判断标准响应里的tool_calls数组中包含query_order且最终回复引用了工具返回结果。如果模型没有生成工具调用检查工具描述是否写清楚、参数格式是否匹配。7.4 长文本与复杂任务测试准备一份较长的业务文档让 Agent 完成“阅读后总结”任务。这类任务可以暴露上下文窗口、摘要策略和 token 消耗问题。如果 Agent 回看前面内容时丢失信息需要引入摘要机制把超长历史压缩后重新送入上下文。7.5 稳定性测试连续发送 20 到 50 个请求观察服务是否出现内存泄漏、超时或返回空结果。一个简单的循环脚本就可以完成import time import requests url http://127.0.0.1:8000/agent/run failed 0 for i in range(20): try: resp requests.post(url, json{ session_id: fstress-{i}, user_input: f第 {i} 个测试问题 }, timeout120) if resp.status_code ! 200: failed 1 except Exception as e: failed 1 print(frequest {i} failed: {e}) time.sleep(1) print(ffailed: {failed}/20)如果失败率过高先排查模型服务是否限流再看 Agent 服务日志中是否出现上下文超长或工具调用异常。8. 接口 API 与批量任务设计8.1 API 接口设计建议Agent 服务对外接口按照业务用途拆分成三类每个接口职责不同便于后续单独扩容。接口路径用途同步调用POST /agent/run单次请求同步返回结果异步任务POST /agent/task提交任务立即返回任务 ID任务查询GET /agent/task/{task_id}查询异步任务状态和结果同步接口适合交互场景异步任务适合耗时较长的批量场景。生产环境建议使用异步任务避免长任务占满 HTTP 连接。8.2 curl 调用示例curl -X POST http://127.0.0.1:8000/agent/run \ -H Content-Type: application/json \ -d { session_id: curl-test, user_input: 总结一下这个项目的核心功能 }8.3 批量任务处理批量任务是企业级场景中最高频的需求。可以按目录扫描输入文件也可以从消息队列拉取任务。这里给出一套目录扫描批量处理的通用思路。inputs/ ├── task_001.txt ├── task_002.txt └── task_003.txt批量消费者逻辑示例# jobs/scanner.py import os import time import json import requests INPUT_DIR inputs OUTPUT_DIR outputs API_URL http://127.0.0.1:8000/agent/run def process_tasks(): for filename in os.listdir(INPUT_DIR): if not filename.endswith(.txt): continue filepath os.path.join(INPUT_DIR, filename) with open(filepath, r, encodingutf-8) as f: content f.read() payload { session_id: fbatch-{filename}, user_input: f请处理以下内容\n{content} } try: resp requests.post(API_URL, jsonpayload, timeout180) result resp.json() output_path os.path.join(OUTPUT_DIR, filename.replace(.txt, .json)) with open(output_path, w, encodingutf-8) as f: json.dump(result, f, ensure_asciiFalse, indent2) print(fprocessed: {filename}) except Exception as e: print(ffailed: {filename}, error: {e}) continue time.sleep(1) if __name__ __main__: process_tasks()批量任务的几个关键点输入和输出分目录管理避免覆盖原文件每条任务都记录成功或失败失败任务先记录日志不立即中断整个队列处理完一批后人工抽样检查结果质量。生产环境建议把任务队列替换为 Redis、RabbitMQ 或消息平台。批量任务量上来之后目录扫描方式会成为瓶颈消息队列可以支持多消费者并行处理和失败重投。8.4 失败重试建议失败类型重试策略模型 API 超时指数退避重试最多 3 次工具调用失败检查工具参数重置上下文后重试批量任务中单条失败记录失败原因跳过并继续处理重复任务按session_id或任务 ID 做幂等控制幂等控制很重要。提交同一批输入时如果因为网络原因重复提交不能让系统重复执行具有副作用的操作比如重复发送消息、重复扣费。每个任务生成唯一 ID在任务表中先查后写。9. 资源占用与性能观察9.1 显存与 GPU 使用观察本地运行大模型时显存占用是第一个需要关注的指标。用以下命令实时监控 GPU 状态watch -n 1 nvidia-smi每次 Agent 请求到来时观察显存变化。单条请求的显存占用通常和模型参数量、上下文长度强相关。实际占用须以本机测试为准不同推理框架、不同量化方式差异很大。如果要稳定支撑并发建议预留 20% 到 30% 的显存余量避免并发请求导致显存溢出。9.2 CPU 推理与 GPU 推理的差异CPU 推理的优势是部署简单、不依赖显卡适合低并发或开发调试。缺点是单次请求延迟明显更高批量任务在高并发下耗时成倍增加。GPU 推理的延迟低、吞吐高但对显存和驱动环境有要求。开发阶段可以全部用 CPU 加小模型跑通流程验证逻辑没问题后再切到 GPU 环境。这样可以在没有显卡的机器上先完成 Harness 架构的调试。9.3 影响性能的主要因素因素影响上下文长度越长推理越慢token 消耗越高工具数量工具描述全部送入模型数量多会占用上下文最大迭代轮数轮数越多耗时越长需设置上限并发请求数并发过高导致排队、超时批量任务大小单批任务越大单次显存占用越高降低占用的常见手段限制上下文长度工具结果做摘要后再送模型调低并发数使用量化模型把超长任务拆成小任务分批处理。如果 API 服务进程跑在 GPU 机器上还要关注显存是否被其他进程占用。启动前用nvidia-smi查看是否已有进程占用显存避免“显存不足”但模型本身没问题的假象。10. 常见问题与排查方法问题现象可能原因排查方式解决方案服务启动后页面打不开端口被占用或服务未启动检查启动日志和端口状态更换端口或重启服务模型调用报连接错误API Base 地址配置错误或模型服务未启动curl 测试模型服务地址修正环境变量先启动模型服务请求超时上下文过长、模型推理慢、工具卡住查看日志中耗时分布缩短上下文限制迭代轮数设超时阈值工具调用格式错误模型输出参数不匹配查看原始模型输出日志调整工具描述增加输出校验和自动修复显存不足模型过大或并发过高查看 nvidia-smi换小模型、降低批量、使用量化版本批量任务卡住单条任务异常未捕获检查任务日志和输出目录给请求加超时加失败自动跳过生成内容质量波动提示词不稳定或上下文污染对比不同输入的完整日志固定系统提示词隔离会话上下文重复执行副作用操作缺少幂等控制检查任务 ID 和请求记录在任务表加唯一约束先查后写端口冲突是本地开发最常见的坑。可以在启动代码中增加端口检测端口被占用时自动提示具体进程而不是直接闪退。批量任务中最怕的是“整个队列一起失败”建议每个文件独立 try-catch失败后把错误写进单独的失败清单等本轮结束后统一重试。11. 企业落地最佳实践与合规提醒第一次接触 Harness 架构时不要一上来就追求大而全。先用小模型、单工具、短上下文跑通主链路再逐步加工具、加记忆、加批量。保留一套最小可运行配置团队其他人接手时能快速启动环境。工程化层面的建议目录分离输入素材、输出结果、日志、配置文件严格分开避免处理过程污染原始数据。版本管理提示词文件纳入 Git 管理每次修改记录变更原因方便效果回退。配置外置模型地址、密钥、端口等用环境变量管理不要硬编码。日志完整每轮 Agent 调用记录用户输入、模型输出、工具调用、耗时、 token 消耗问题排查有据可依。批量失败重试单条失败不阻塞队列统一记录失败原因结束后人工处理。接口鉴权Agent 服务内部使用时至少加简单的 Token 校验不能裸奔在公网。内容复核AI 生成内容在商用前要做人工抽查特别是涉及价格、政策、医疗、法律等敏感领域。工具权限最小化Agent 能调用的工具按需开放只读工具不给写权限写操作走审批。合规方面要特别注意涉及真实人物声音、人脸、肖像时必须有明确授权使用企业内部数据训练或接入外部模型时先确认数据出域是否合规Agent 自动发送消息、自动下单、自动发布内容必须经过人工确认环节。不管架构做得多么完善最后一道把关始终是人。12. 总结与下一步Harness 架构最值得尝试的点是把混乱的 Agent 开发过程规范成一条可维护的流水线。模型负责智能Harness 负责稳定。先验证的不应该是复杂规划能力而是工具调用是否稳定、批量任务是否可靠、日志是否能支撑问题回溯。最容易踩的坑有两个一个是上下文越拖越长导致延迟和成本双涨另一个是工具权限控制不到位Agent 在测试环境里执行了不该执行的操作。前者靠上下文摘要和长度阈值解决后者靠权限白名单和人工审批兜底。后续可以扩展的方向很明确把内存缓存换成向量数据库让 Agent 具备长期记忆把批量任务从目录扫描迁移到消息队列支持水平扩容和多消费者并行增加独立的评估集合每次修改提示词或工具逻辑后自动跑回归测试。到这一步就是一个具备工程雏形的企业级 Agent 服务了。 
长河君 · 2026/9/7 13:07:54 · 阅读 1.2k Harness架构：从AI Agent开发到企业级稳定落地的工程化指南
大模型跑通不难难的是把它做成一个稳定、可控、能接业务的企业级服务。这次我们来看 Harness 架构它不是什么新框架而是一套把“模型调用、工具执行、上下文管理、失败重试、日志追踪”统一收口的工程化模式。结合最近社区里频繁提到的 DeepAgent、Agent 开发路线这篇文章会从架构概念讲到企业级项目落地再给出一套可以在本机验证的部署、测试和排错流程。如果只关心单轮问答确实用不到 Harness但一旦涉及多工具编排、批量任务、接口 API 和长期运行Harness 的价值会非常明显。本文不打算讲营销层面的“吊打付费”而是把 Harness 在 AI 大模型和 Agent 开发里到底怎么理解、怎么落地讲清楚。文章会先给出核心能力速览然后用一套可以照做的项目结构演示如何搭建 Agent 服务最后覆盖接口调用、批量任务、资源占用观察和常见问题排查。适合准备做 AI Agent 应用开发、想从 Demo 走向工程化的读者。1. Harness 架构核心能力速览能力项说明架构定位Agent 执行控制层工程化模式不是某个单一开源项目核心功能提示词模板管理、工具注册与调度、上下文管理、执行循环、失败重试、日志追踪典型应用企业知识库问答、自动化报表、智能客服、批量文档处理、多步骤任务编排运行环境CPU 可运行推理加速推荐 GPU具体显存按模型版本测试启动方式Python 命令行启动 / FastAPI 服务启动 / Celery 批量任务节点接口能力通过 HTTP API 对外暴露 Agent 调用能力批量任务支持任务队列目录扫描批量处理或消息队列分发是否支持一键启动依赖项目封装方式建议使用脚本加虚拟环境适合人群后端开发者、AI 应用工程师、算法工程师、想要规范 Agent 项目的团队合规边界自动操作外部系统前必须授权生成内容需人工复核数据隐私按企业规范执行从材料来看Harness 在社区里常与 DeepAgent、Agent 框架放在一起讨论本质上是解决“模型输出不可控、工具调用不稳定、长任务容易断”的问题。架构本身不绑定特定大模型适合作为 AI 大模型应用开发的通用底座。2. 适用场景与使用边界Harness 架构最适合的是有一定流程规律、需要模型反复调用工具的任务。比如一个客服机器人需要先查订单、再查物流、最后生成回复或者一个数据分析 Agent需要按用户问题选择 SQL 查询、Python 计算、图表生成。这类场景中模型负责判断下一步Harness 负责保证每一步都能执行、可重试、可追踪。它在这些场景里效果明显多步骤任务需要多次模型推理和工具调用而不是一次问答。批量处理大量输入需要走同一套 Agent 流程。接口化交付业务方需要把 Agent 能力封装成 HTTP API。可控性要求高生成结果需要记录完整调用链方便回查。不适合的场景同样明确。单次短问答没必要引入 Harness直接用模型 API 更省事需要毫秒级响应的在线接口也不适合把 Agent 编排放在同步链路里另外如果业务本身没有明确工具边界Agent 自由调用工具反而容易出问题。安全边界必须强调。Agent 一旦具备工具调用能力就意味着它能操作文件、访问数据库、调用第三方接口。企业内部使用时需要对工具做白名单控制给每个工具设置权限等级涉及用户隐私数据时只允许在合规环境内处理自动发送消息、自动下单、自动发布内容等操作必须经过人工审批环节。所有生成内容在正式上线前都应该做抽样复核。3. Harness 架构到底在解决什么问题把模型直接接到业务里通常会遇到几个问题。第一个是提示词散落。每个业务场景都要写一段很长的系统提示词散落在代码里改一次要动代码。Harness 把提示词模板独立成管理模块按场景、按版本维护。第二个是工具调用不稳定。模型输出工具调用参数时经常出现格式错误、参数缺失、调用了不存在的工具。Harness 在模型和工具之间加一层校验先解析模型输出再匹配工具定义最后执行工具并记录结果。第三个是上下文管理困难。多轮对话中历史消息、工具返回结果、临时变量混在一起容易把上下文撑爆。Harness 负责把工具结果整理成结构化内容决定哪些进上下文、哪些截断、哪些摘要。第四个是失败重试缺失。工具请求第三方超时是直接返回错误还是重试Harness 统一处理超时、限流、重试避免业务代码到处写 try-catch。第五个是无法观测。模型在做什么、调了哪些工具、花了多长时间、消耗了多少 token如果没有统一日志排查问题非常痛苦。Harness 在每次执行循环里记录完整轨迹。从架构层次看一个典型的 Agent Harness 包含这么几层层级职责关键模块交互层接收用户输入返回最终结果WebUI、HTTP API、命令行执行控制层维护 Agent 主循环决定下一步动作Harness 核心、状态机、循环调度模型接入层对接不同大模型统一输入输出OpenAI 兼容接口、本地模型服务工具层注册和执行业务能力搜索、数据库、计算、文档解析等记忆层管理短期和长期上下文会话缓存、向量库、摘要器可观测层日志、追踪、评估调用链日志、指标采集、错误报告DeepAgent 在社区语境里通常指具备深度任务规划能力的 Agent核心是让执行控制层支持更复杂的任务拆解、长周期执行、反思和校验。它的工程实现依然依赖 Harness 这套底座只是在规划算法和记忆机制上做得更深。4. 企业级 Agent 项目目录设计与模块拆分一个可以在本机跑起来、又能逐步演进成企业级服务的 Agent 项目目录结构建议这样拆分。这里以 Python 为例实际项目可以按团队习惯调整。agent-project/ ├── app.py # FastAPI 入口 ├── config.py # 全局配置 ├── requirements.txt # 依赖清单 ├── prompts/ # 提示词模板目录 │ ├── system_planner.txt │ ├── system_tool_call.txt │ └── summary.txt ├── tools/ # 工具定义 │ ├── __init__.py │ ├── registry.py # 工具注册中心 │ ├── web_search.py # 搜索工具 │ ├── sql_runner.py # 数据库工具 │ └── file_reader.py # 文件读取工具 ├── harness/ │ ├── __init__.py │ ├── loop.py # Agent 主循环 │ ├── context.py # 上下文管理 │ ├── parser.py # 模型输出解析 │ └── retry.py # 重试策略 ├── memory/ │ ├── __init__.py │ └── buffer.py # 会话缓冲 ├── api/ │ ├── __init__.py │ └── routes.py # HTTP 路由 ├── jobs/ # 批量任务 │ ├── __init__.py │ ├── consumer.py # 任务消费者 │ └── scanner.py # 目录扫描 ├── logs/ # 运行日志 ├── inputs/ # 批量输入素材 └── outputs/ # 批量输出结果模块拆分的核心原则提示词、工具、执行逻辑、接口、批量任务互不耦合。业务方新增一个能力时只需要在tools/下新增一个工具并注册调整模型行为时只改prompts/下的模板文件要接新前端不需要改动 Harness 核心。工具注册中心是其中一个关键模块。每个工具都声明名称、描述、参数 URL、权限级别模型根据这些描述决定是否调用。# tools/registry.py from dataclasses import dataclass, field from typing import Callable, Any dataclass class Tool: name: str description: str parameters: dict permission: str read handler: Callable[..., Any] None class ToolRegistry: def __init__(self): self._tools {} def register(self, tool: Tool): self._tools[tool.name] tool def get(self, name: str) - Tool: return self._tools.get(name) def list_tools(self): return [ { name: t.name, description: t.description, parameters: t.parameters, } for t in self._tools.values() ] registry ToolRegistry()这个示例展示了工具注册的基本思路。实际项目中可以在registry.register装饰器里完成工具声明每增加一个业务能力就注册一次Harness 主循环从注册中心获取工具列表并生成给模型的工具描述。5. Harness 本地部署环境准备Harness 架构本身不依赖特定硬件核心依赖是大模型推理服务。如果使用云端模型 API常规开发机即可如果全部本地部署需要按模型规模准备 GPU。更稳妥的判断是先跑通一个小尺寸模型再评估是否升级硬件。环境准备清单如下检查项建议操作系统Linux 服务器优先Windows/macOS 可用于本地开发Python 版本3.10 或 3.11依赖管理pip virtualenv 或 conda大模型服务OpenAI 兼容 API 或本地 Ollama / vLLMCUDA使用本地 GPU 时安装对应驱动和 CUDA磁盘空间依赖包 2GB 左右本地模型按实际大小预留端口规划API 服务默认建议 8000避免冲突Python 虚拟环境创建示例python3 -m venv venv source venv/bin/activate pip install --upgrade pip pip install fastapi uvicorn pydantic requests openai python-dotenv如果使用本地模型需要提前启动模型推理服务。Ollama 启动方式非常直接ollama pull qwen2.5:7b ollama serve模型服务启动后用curl http://127.0.0.1:11434/api/tags验证是否就绪。如果返回模型列表说明推理服务可访问。显存占用以实际模型版本和推理参数为准先跑一次短问题观察nvidia-smi的占用再决定是否增大批量。6. 安装部署与启动方式先准备环境变量文件配置模型接入地址和 API Key。这里以 OpenAI 兼容接口为例# .env 示例 MODEL_NAMEqwen2.5:7b OPENAI_API_BASEhttp://127.0.0.1:11434/v1 OPENAI_API_KEYollama APP_HOST127.0.0.1 APP_PORT8000如果使用云端模型把OPENAI_API_BASE换成云厂商兼容地址OPENAI_API_KEY换成真实密钥。注意不要把密钥提交到代码仓库。接着启动 Agent API 服务。以 FastAPI 为例入口如下# app.py import os from fastapi import FastAPI from pydantic import BaseModel app FastAPI(titleAgent Harness Service) class AgentRequest(BaseModel): session_id: str default user_input: str stream: bool False class AgentResponse(BaseModel): session_id: str output: str tool_calls: list trace_id: str app.post(/agent/run, response_modelAgentResponse) async def agent_run(req: AgentRequest): # 实际项目在这里调用 harness.loop.run() # 返回结果应包含最终回复、工具调用记录和追踪 ID return AgentResponse( session_idreq.session_id, outputAgent 执行结果, tool_calls[], trace_idtrace-001 ) if __name__ __main__: import uvicorn uvicorn.run(app, hostos.getenv(APP_HOST, 127.0.0.1), portint(os.getenv(APP_PORT, 8000)))启动命令source venv/bin/activate python app.py启动后访问http://127.0.0.1:8000/docs如果能看到 Swagger 文档说明服务启动成功。第一次启动建议先关闭其他占用 8000 端口的进程避免端口冲突。如果需要在后台运行服务并保留日志nohup python app.py logs/app.log 21 echo $! logs/app.pid停止服务时先读取 PID 再杀掉进程避免出现进程残留。7. 功能测试与效果验证服务启动后按照“先单轮、再多轮、再工具调用、再批量”的顺序进行验证。每个环节都记录输入、输出、耗时和显存占用作为后续调优的依据。7.1 单轮问答测试第一步先验证模型连通性和基础回答质量。import requests url http://127.0.0.1:8000/agent/run payload { session_id: test-basic, user_input: 用一句话解释什么是 Harness 架构 } response requests.post(url, jsonpayload, timeout120) print(response.json())预期结果是返回一段通顺的中文解释。如果模型服务未配置好这里通常会报连接错误或超时优先检查OPENAI_API_BASE是否可达。7.2 多轮对话测试多轮测试主要验证会话上下文是否被正确保存。payload_1 { session_id: test-multi, user_input: 我要做企业知识库问答系统先帮我列出三个核心模块 } response_1 requests.post(url, jsonpayload_1, timeout120) print(response_1.json()[output]) payload_2 { session_id: test-multi, user_input: 第一个模块的具体实现步骤是什么 } response_2 requests.post(url, jsonpayload_2, timeout120) print(response_2.json()[output])预期结果是第二轮回答能关联第一轮提到的三个模块。如果第二轮回答跑题检查上下文管理模块是否把历史消息传给了模型。7.3 工具调用测试工具调用是 Harness 架构的重头戏。在工具注册中心添加一个模拟查询工具然后让 Agent 调用它。def mock_order_query(order_id: str): return {order_id: order_id, status: 已发货, logistics: 顺丰速运} registry.register(Tool( namequery_order, description根据订单号查询订单状态, parameters{order_id: {type: string, description: 订单号}}, permissionread, handlermock_order_query ))测试输入{ session_id: test-tool, user_input: 帮我查一下订单 20240001 现在什么状态 }预期结果是 Agent 先解析出需要调用query_order工具然后返回订单状态。判断标准响应里的tool_calls数组中包含query_order且最终回复引用了工具返回结果。如果模型没有生成工具调用检查工具描述是否写清楚、参数格式是否匹配。7.4 长文本与复杂任务测试准备一份较长的业务文档让 Agent 完成“阅读后总结”任务。这类任务可以暴露上下文窗口、摘要策略和 token 消耗问题。如果 Agent 回看前面内容时丢失信息需要引入摘要机制把超长历史压缩后重新送入上下文。7.5 稳定性测试连续发送 20 到 50 个请求观察服务是否出现内存泄漏、超时或返回空结果。一个简单的循环脚本就可以完成import time import requests url http://127.0.0.1:8000/agent/run failed 0 for i in range(20): try: resp requests.post(url, json{ session_id: fstress-{i}, user_input: f第 {i} 个测试问题 }, timeout120) if resp.status_code ! 200: failed 1 except Exception as e: failed 1 print(frequest {i} failed: {e}) time.sleep(1) print(ffailed: {failed}/20)如果失败率过高先排查模型服务是否限流再看 Agent 服务日志中是否出现上下文超长或工具调用异常。8. 接口 API 与批量任务设计8.1 API 接口设计建议Agent 服务对外接口按照业务用途拆分成三类每个接口职责不同便于后续单独扩容。接口路径用途同步调用POST /agent/run单次请求同步返回结果异步任务POST /agent/task提交任务立即返回任务 ID任务查询GET /agent/task/{task_id}查询异步任务状态和结果同步接口适合交互场景异步任务适合耗时较长的批量场景。生产环境建议使用异步任务避免长任务占满 HTTP 连接。8.2 curl 调用示例curl -X POST http://127.0.0.1:8000/agent/run \ -H Content-Type: application/json \ -d { session_id: curl-test, user_input: 总结一下这个项目的核心功能 }8.3 批量任务处理批量任务是企业级场景中最高频的需求。可以按目录扫描输入文件也可以从消息队列拉取任务。这里给出一套目录扫描批量处理的通用思路。inputs/ ├── task_001.txt ├── task_002.txt └── task_003.txt批量消费者逻辑示例# jobs/scanner.py import os import time import json import requests INPUT_DIR inputs OUTPUT_DIR outputs API_URL http://127.0.0.1:8000/agent/run def process_tasks(): for filename in os.listdir(INPUT_DIR): if not filename.endswith(.txt): continue filepath os.path.join(INPUT_DIR, filename) with open(filepath, r, encodingutf-8) as f: content f.read() payload { session_id: fbatch-{filename}, user_input: f请处理以下内容\n{content} } try: resp requests.post(API_URL, jsonpayload, timeout180) result resp.json() output_path os.path.join(OUTPUT_DIR, filename.replace(.txt, .json)) with open(output_path, w, encodingutf-8) as f: json.dump(result, f, ensure_asciiFalse, indent2) print(fprocessed: {filename}) except Exception as e: print(ffailed: {filename}, error: {e}) continue time.sleep(1) if __name__ __main__: process_tasks()批量任务的几个关键点输入和输出分目录管理避免覆盖原文件每条任务都记录成功或失败失败任务先记录日志不立即中断整个队列处理完一批后人工抽样检查结果质量。生产环境建议把任务队列替换为 Redis、RabbitMQ 或消息平台。批量任务量上来之后目录扫描方式会成为瓶颈消息队列可以支持多消费者并行处理和失败重投。8.4 失败重试建议失败类型重试策略模型 API 超时指数退避重试最多 3 次工具调用失败检查工具参数重置上下文后重试批量任务中单条失败记录失败原因跳过并继续处理重复任务按session_id或任务 ID 做幂等控制幂等控制很重要。提交同一批输入时如果因为网络原因重复提交不能让系统重复执行具有副作用的操作比如重复发送消息、重复扣费。每个任务生成唯一 ID在任务表中先查后写。9. 资源占用与性能观察9.1 显存与 GPU 使用观察本地运行大模型时显存占用是第一个需要关注的指标。用以下命令实时监控 GPU 状态watch -n 1 nvidia-smi每次 Agent 请求到来时观察显存变化。单条请求的显存占用通常和模型参数量、上下文长度强相关。实际占用须以本机测试为准不同推理框架、不同量化方式差异很大。如果要稳定支撑并发建议预留 20% 到 30% 的显存余量避免并发请求导致显存溢出。9.2 CPU 推理与 GPU 推理的差异CPU 推理的优势是部署简单、不依赖显卡适合低并发或开发调试。缺点是单次请求延迟明显更高批量任务在高并发下耗时成倍增加。GPU 推理的延迟低、吞吐高但对显存和驱动环境有要求。开发阶段可以全部用 CPU 加小模型跑通流程验证逻辑没问题后再切到 GPU 环境。这样可以在没有显卡的机器上先完成 Harness 架构的调试。9.3 影响性能的主要因素因素影响上下文长度越长推理越慢token 消耗越高工具数量工具描述全部送入模型数量多会占用上下文最大迭代轮数轮数越多耗时越长需设置上限并发请求数并发过高导致排队、超时批量任务大小单批任务越大单次显存占用越高降低占用的常见手段限制上下文长度工具结果做摘要后再送模型调低并发数使用量化模型把超长任务拆成小任务分批处理。如果 API 服务进程跑在 GPU 机器上还要关注显存是否被其他进程占用。启动前用nvidia-smi查看是否已有进程占用显存避免“显存不足”但模型本身没问题的假象。10. 常见问题与排查方法问题现象可能原因排查方式解决方案服务启动后页面打不开端口被占用或服务未启动检查启动日志和端口状态更换端口或重启服务模型调用报连接错误API Base 地址配置错误或模型服务未启动curl 测试模型服务地址修正环境变量先启动模型服务请求超时上下文过长、模型推理慢、工具卡住查看日志中耗时分布缩短上下文限制迭代轮数设超时阈值工具调用格式错误模型输出参数不匹配查看原始模型输出日志调整工具描述增加输出校验和自动修复显存不足模型过大或并发过高查看 nvidia-smi换小模型、降低批量、使用量化版本批量任务卡住单条任务异常未捕获检查任务日志和输出目录给请求加超时加失败自动跳过生成内容质量波动提示词不稳定或上下文污染对比不同输入的完整日志固定系统提示词隔离会话上下文重复执行副作用操作缺少幂等控制检查任务 ID 和请求记录在任务表加唯一约束先查后写端口冲突是本地开发最常见的坑。可以在启动代码中增加端口检测端口被占用时自动提示具体进程而不是直接闪退。批量任务中最怕的是“整个队列一起失败”建议每个文件独立 try-catch失败后把错误写进单独的失败清单等本轮结束后统一重试。11. 企业落地最佳实践与合规提醒第一次接触 Harness 架构时不要一上来就追求大而全。先用小模型、单工具、短上下文跑通主链路再逐步加工具、加记忆、加批量。保留一套最小可运行配置团队其他人接手时能快速启动环境。工程化层面的建议目录分离输入素材、输出结果、日志、配置文件严格分开避免处理过程污染原始数据。版本管理提示词文件纳入 Git 管理每次修改记录变更原因方便效果回退。配置外置模型地址、密钥、端口等用环境变量管理不要硬编码。日志完整每轮 Agent 调用记录用户输入、模型输出、工具调用、耗时、 token 消耗问题排查有据可依。批量失败重试单条失败不阻塞队列统一记录失败原因结束后人工处理。接口鉴权Agent 服务内部使用时至少加简单的 Token 校验不能裸奔在公网。内容复核AI 生成内容在商用前要做人工抽查特别是涉及价格、政策、医疗、法律等敏感领域。工具权限最小化Agent 能调用的工具按需开放只读工具不给写权限写操作走审批。合规方面要特别注意涉及真实人物声音、人脸、肖像时必须有明确授权使用企业内部数据训练或接入外部模型时先确认数据出域是否合规Agent 自动发送消息、自动下单、自动发布内容必须经过人工确认环节。不管架构做得多么完善最后一道把关始终是人。12. 总结与下一步Harness 架构最值得尝试的点是把混乱的 Agent 开发过程规范成一条可维护的流水线。模型负责智能Harness 负责稳定。先验证的不应该是复杂规划能力而是工具调用是否稳定、批量任务是否可靠、日志是否能支撑问题回溯。最容易踩的坑有两个一个是上下文越拖越长导致延迟和成本双涨另一个是工具权限控制不到位Agent 在测试环境里执行了不该执行的操作。前者靠上下文摘要和长度阈值解决后者靠权限白名单和人工审批兜底。后续可以扩展的方向很明确把内存缓存换成向量数据库让 Agent 具备长期记忆把批量任务从目录扫描迁移到消息队列支持水平扩容和多消费者并行增加独立的评估集合每次修改提示词或工具逻辑后自动跑回归测试。到这一步就是一个具备工程雏形的企业级 Agent 服务了。