在 AI Agent 工程里Harness 架构并不是某个框架的专属名词也不是界面工具而是指位于大模型与业务逻辑之间的那一层编排脚手架。它的核心价值在于把模型调用、工具注册、多轮上下文、终止条件和错误处理统一到一个可控流程中让 Agent 不再只是“提示词 API”的临时脚本而是一个可测试、可运维、可扩展的工程系统。这篇文章会围绕 DeepSeek 模型接口从零实现一个最小可运行的 Agent Harness再逐步补充企业级开发需要关注的配置、日志、限流和排查手段。无论你是刚开始接触 AI 大模型应用开发还是已经在生产环境维护 Agent 服务都可以把这里的思路作为落地基础。1. 理解 Harness 架构Agent 工程里到底多出来的是什么很多新手会误以为 Agent 就是“大模型 工具调用”写几个函数让模型根据用户问题选择执行即可。真正开始工程化之后会发现模型只负责“决策”不负责“执行”。它返回的是一段工具调用指令而真正去查数据库、调接口、写文件、做异常处理的工作需要一个外围框架完成。这个框架就是 Harness。1.1 先区分 Agent、Harness 和模型调用Agent 是逻辑实体表示一个能够感知环境、做出决策并调用工具完成任务的人工智能体。模型调用是单次与大模型的交互。Harness 则是承载 Agent 运行的环境它知道如何发起模型请求如何把工具结果重新传给模型如何判断任务是否结束以及每一步失败时该做什么。用一句话概括Agent 是“做什么”Harness 是“怎么跑起来”。在社区讨论中经常出现的 deepseek harness、codex harness、agent harness 等名称本质上都是围绕某个模型或某个场景搭建的一套可复用执行脚手架。不同项目的实现细节差异很大但核心结构一致模型端口、工具注册表、消息上下文、循环控制器、终止条件。1.2 Harness 的核心组件一个最小 Harness 至少包含以下部分组件职责示例模型客户端封装对 DeepSeek 等模型的 API 调用OpenAI SDK、requests 封装消息上下文保存系统提示词、用户输入、助手回复、工具结果messages 列表工具注册表描述可用工具并映射到实际函数schema 列表 函数字典循环控制器交替执行“模型决策”和“工具执行”for 循环或 while 循环终止条件决定何时停止迭代并返回结果无 tool_calls、达到最大步数异常处理处理解析失败、超时、服务不可用try/except、重试、降级这些组件看起来简单但每项单独拎出来都有工程细节。例如消息上下文不能无限增长工具注册表的 schema 必须与函数签名严格一致终止条件不能只依赖模型输出还要考虑最大迭代次数的保护。1.3 什么时候需要自己写 Harness如果只是做一个演示 Demo让模型回答几个问题不涉及外部工具那不需要 Harness直接调用模型接口即可。一旦出现以下情况自研或引入 Harness 就有必要模型需要读取数据库、调用订单系统、操作文件一次任务需要多轮工具调用才能完成需要对每一次模型请求做日志审计需要在不同环境使用不同模型服务需要控制成本、超时和并发。自研最小 Harness 的好处是能清楚理解每一步在做什么。引入成熟框架的好处是少写重复代码。两者不冲突建议先自研一次再决定是否需要使用框架。2. 环境准备用 DeepSeek 模型后端跑通第一行代码在实现完整 Harness 之前先把模型接口跑通。DeepSeek 提供 OpenAI 兼容的 API 接口因此可以直接使用 openai Python SDK。这里要避免一个常见误区不要一上来就封装一堆类先写一个最简调用确认 API Key、模型名、网络和依赖都正确。2.1 准备 API Key 和依赖先在服务商控制台创建 API Key然后安装依赖mkdir deepseek-agent-harness cd deepseek-agent-harness python -m venv venv source venv/bin/activate pip install openai python-dotenv这里使用python-dotenv是为了把密钥放在环境变量文件中避免写死在代码里。项目中创建.env文件DEEPSEEK_API_KEYsk-xxxx DEEPSEEK_BASE_URLhttps://api.deepseek.com DEEPSEEK_MODELdeepseek-chat不同服务商提供的 base_url 和模型名称可能不同落地前要先确认。密钥不要提交到 Git 仓库建议同时在.gitignore中加入.env。2.2 项目结构与配置保持目录清晰方便后续扩展deepseek-agent-harness/ ├── .env ├── requirements.txt ├── harness/ │ ├── __init__.py │ ├── client.py │ ├── tools.py │ ├── agent.py │ └── config.py └── main.py这是常见的最小工程结构。client.py负责创建模型客户端tools.py维护工具 schema 和函数实现agent.py实现主循环config.py读取环境变量。2.3 验证模型接口可用先写一个验证脚本import os from dotenv import load_dotenv from openai import OpenAI load_dotenv() client OpenAI( api_keyos.getenv(DEEPSEEK_API_KEY), base_urlos.getenv(DEEPSEEK_BASE_URL), ) resp client.chat.completions.create( modelos.getenv(DEEPSEEK_MODEL), messages[ {role: system, content: 你是一个测试助手回答请简短。}, {role: user, content: 11 等于多少}, ], ) print(resp.choices[0].message.content)运行后能看到模型返回的文本。这一步通过后再进入 Harness 主循环开发。注意不要只看程序是否成功启动要确认模型确实返回了预期内容。API Key 权限不足、base_url 配错、模型名不存在都会在这个环节暴露。3. 从零实现一个最小 DeepSeek Agent Harness现在开始实现核心代码。为了便于理解用一个具体业务场景电商订单查询。用户输入订单号Agent 决定调用get_order_status工具工具返回状态模型最后生成回答。3.1 工具注册表让模型知道有什么可用函数实现def get_order_status(order_id: str) - str: 根据订单号查询订单状态。 data { A1001: 已发货, A1002: 待支付, A1003: 已取消, } return data.get(order_id, 订单不存在)为了让模型能知道这个工具的存在需要按 Function Calling 协议提供 schemaTOOLS_SCHEMA [ { type: function, function: { name: get_order_status, description: 根据订单号查询订单状态, parameters: { type: object, properties: { order_id: { type: string, description: 订单号例如 A1001 } }, required: [order_id] }, }, } ] TOOL_FUNCTIONS { get_order_status: get_order_status, }这里有两个关键点。第一description必须写清楚工具的作用和参数含义模型会依据描述决定是否调用。第二parameters中的字段名要和函数参数名完全一致否则解析参数后会调用失败。3.2 Agent 主循环消息组装、工具调用、结果回填主循环是 Harness 的心脏。import json from openai import OpenAI from dotenv import load_dotenv load_dotenv() client OpenAI( api_keyos.getenv(DEEPSEEK_API_KEY), base_urlos.getenv(DEEPSEEK_BASE_URL), ) MODEL_NAME os.getenv(DEEPSEEK_MODEL) SYSTEM_PROMPT ( 你是电商客服助手。 查询订单状态时必须使用 get_order_status 工具。 不要编造订单信息。 ) def run_agent(user_input: str, max_iterations: int 5): messages [ {role: system, content: SYSTEM_PROMPT}, {role: user, content: user_input}, ] for step in range(max_iterations): resp client.chat.completions.create( modelMODEL_NAME, messagesmessages, toolsTOOLS_SCHEMA, tool_choiceauto, ) msg resp.choices[0].message messages.append(msg.model_dump()) if not msg.tool_calls: return msg.content for tool_call in msg.tool_calls: fn_name tool_call.function.name fn_args json.loads(tool_call.function.arguments or {}) fn TOOL_FUNCTIONS.get(fn_name) if fn is None: result f未知工具: {fn_name} else: result fn(**fn_args) messages.append({ role: tool, tool_call_id: tool_call.id, name: fn_name, content: json.dumps({result: result}, ensure_asciiFalse), }) return 达到最大迭代次数未得到最终结果这段代码的核心流程是把 system 和 user 消息传给模型模型返回 assistant 消息可能包含tool_calls如果模型没有申请工具调用说明任务已经完成直接返回内容如果模型申请了工具调用执行对应函数把每个工具结果以role: tool的消息追加到上下文进入下一轮循环让模型基于工具结果继续回答。msg.model_dump()是一个关键操作。它把 assistant 消息完整保留在消息列表中包括content、tool_calls等字段。如果只保存content下一轮模型会丢失自己发出的工具调用请求无法正确组织后续消息。3.3 完整示例查询订单状态在主入口中加入运行代码if __name__ __main__: answer run_agent(请帮我查一下订单 A1001 的状态) print(answer)运行python main.py预期输出类似订单 A1001 当前状态是已发货。3.4 运行验证与预期输出验证时不要只跑一个正常用例。至少要验证三类输入输入预期行为关注点正常订单号模型调用工具返回状态工具调用链路是否完整不存在的订单号模型调用工具返回“订单不存在”工具结果是否能被模型理解与工具无关的问题模型不调用工具直接回答终止条件是否正确还可以打印中间消息观察每一轮模型输出和工具结果answer, messages run_agent_with_messages(请帮我查一下订单 A1001 的状态) for m in messages: print(m)注意工具结果必须如实回填给模型不能在工具函数中打印替代返回值否则模型拿不到数据无法生成最终回答。4. 关键机制详解Harness 为什么能限制模型行为初看主循环只有几十行代码但背后涉及多个机制。理解了这些机制才能排查真实问题。4.1 Function Calling 的协议与约束DeepSeek API 支持 OpenAI 风格的 function calling。模型不会直接调用 Python 函数它只会输出一个结构化的 JSON 指令例如{ name: get_order_status, arguments: {\order_id\: \A1001\} }真正执行函数的是 Harness 代码。Harness 负责把 JSON 字符串解析成参数调用对应函数然后把结果返回给模型。这个设计的好处是模型不需要了解底层实现工具调用可以被记录、审计、限制工具执行失败时Harness 可以返回错误信息让模型重新尝试。约束在于工具 schema 必须与函数签名严格对应。如果模型生成的参数类型不正确Harness 要么严格解析抛错要么把错误信息回传给模型让它修正。4.2 上下文窗口与消息管理每次迭代都会把 assistant 消息和 tool 消息追加到messages中。多轮工具调用会让上下文快速增长。假设一个工具结果有 500 token执行 10 次工具调用仅工具结果就占 5000 token。如果再叠加历史对话、系统提示词很快就会接近模型上下文窗口上限。在企业级 Harness 中需要增加上下文裁剪策略丢弃最旧的对话消息对长工具结果做摘要把关键信息提取后压缩存储超过阈值时直接拒绝继续执行。这些策略要结合实际业务选择不能简单截断否则可能丢失任务上下文。4.3 终止条件与最大迭代次数主循环的终止条件有两个模型不再返回tool_calls或达到max_iterations。第一个条件看似正常但要注意一个隐藏问题模型可能一直生成工具调用指令但参数始终错误导致任务永远无法完成。这时候第二个条件就是安全网。比如设置最大迭代次数为 5超过后返回“未得到最终结果”避免无限消耗 token 和 API 费用。还有一类情况需要人工干预工具执行结果提示当前用户无权限、订单不存在、余额不足。这时 Harness 不应该无脑继续调用而应该把业务状态转换为终止条件。4.4 重要参数选型表实际开发中chat.completions.create的参数会影响模型行为。参数含义常见取值调大影响调小影响temperature随机性0 到 2常见 0.2回答更发散回答更确定top_p核采样概率0.8 到 1.0候选词更多候选词更少max_tokens单次最大输出 token按任务设置输出更长成本更高输出受限可能截断presence_penalty根据话题重复程度惩罚-2 到 2鼓励新话题更集中在已有话题frequency_penalty根据词频惩罚-2 到 2减少重复更容易重复在 Agent Harness 中工具调用场景通常希望模型更稳定所以temperature建议设置为 0.2 左右max_tokens根据工具结果长度调整。如果一次生成包含多个工具调用max_tokens设置过小会导致输出被截断JSON 不完整。5. 从 Demo 到生产企业级 Harness 需要补哪些能力最小 Harness 能跑通但不等于能上线。生产环境面临着配置管理、可观测性、异常处理、并发控制等全新问题。5.1 配置外置化与多环境隔离开发环境、测试环境、生产环境通常使用不同的 API Key、模型服务、数据库、超时时间。不要把配置硬编码在代码里。推荐做法是使用环境变量或配置中心# 开发环境 .env.development DEEPSEEK_API_KEYsk-dev DEEPSEEK_BASE_URLhttps://api.deepseek.com DEEPSEEK_MODELdeepseek-chat LOG_LEVELdebug # 生产环境 .env.production DEEPSEEK_API_KEYsk-prod DEEPSEEK_BASE_URLhttps://api.deepseek.com DEEPSEEK_MODELdeepseek-chat LOG_LEVELinfo同时要建立配置校验机制。应用启动时检查必填配置是否存在不要等到第一次调用模型才报错。5.2 可观测性日志、追踪、审计生产环境每次 Agent 执行都涉及多次模型调用和工具调用。没有日志问题排查会非常痛苦。推荐输出结构化日志至少包含{ timestamp: 2025-01-01T12:00:00Z, trace_id: abc123, user_input: 请查询订单 A1001, model: deepseek-chat, step: 1, tool_name: get_order_status, tool_args: {order_id: A1001}, tool_result: 已发货, latency_ms: 340 }如果团队已有 OpenTelemetry 等追踪体系可以把每次模型调用和工具执行作为 span 上报。没有的话至少要在日志中保留trace_id把同一次 Agent 执行的所有日志串联起来。5.3 异常处理、重试与降级模型接口可能出现超时、限流、返回空响应等异常。在 Agent 主循环中需要区分临时错误和确定性错误。临时错误包括网络超时服务端 429 限流服务端 5xx。这些错误可以使用指数退避重试。常见的 Python 库是tenacityfrom tenacity import retry, stop_after_attempt, wait_exponential retry( stopstop_after_attempt(3), waitwait_exponential(multiplier1, min2, max10), reraiseTrue, ) def chat_with_retry(**kwargs): return client.chat.completions.create(**kwargs)确定性错误则不应该重试比如工具 schema 配置错误API Key 权限不足请求参数非法。错误处理还要考虑“模型已经发了工具调用但工具执行失败”的场景。这时不要直接终止整个 Harness而应该把错误信息作为工具结果回传给模型让模型决定是否修正参数或结束任务。5.4 并发、限流与成本控制Agent 服务一旦上线会有多个用户同时调用。每个用户可能触发多次模型请求并发量会被放大。生产环境至少要做好三件事为每个用户或每个会话设置最大并发数为模型调用侧设置 QPS 限流为单次任务设置 token 预算和最大迭代次数。成本控制方面可以使用以下策略策略说明最大迭代次数防止死循环调用token 预算统计每次请求的 token超过阈值终止缓存对相同问题或相同工具结果做缓存轻量模型简单任务走小模型复杂任务走大模型6. 常见问题排查现象、原因、处理Harness 出了问题最容易踩坑的是以下几类。6.1 模型返回空响应或执行器超时现象调用模型接口后content为空或者报出类似the agent execution provider did not respond in time的提示。可能原因模型服务响应时间超过执行器超时阈值请求没有拿到合法响应但循环仍在等待max_tokens设置过小输出被截断模型服务当前负载较高或被限流。检查方式查看模型 API 返回的完整响应体确认finish_reason打印请求耗时确认是否接近超时阈值查看服务端日志和限流返回状态码尝试用不带工具调用的最小请求复现。处理建议增加客户端超时时间对模型调用增加重试改为流式输出降低首包等待时间设置更明确的空响应拦截逻辑避免无限等待。6.2 工具调用参数解析失败现象json.loads(tool_call.function.arguments)抛异常或函数调用时报缺少参数。可能原因模型生成 JSON 不完整parametersschema 描述与函数签名不一致max_tokens太小导致 arguments 被截断模型不了解参数类型传入了错误结构。处理建议在解析参数时使用try/except把解析错误返回给模型检查工具 schema 的required和字段类型给参数增加更明确的 description工具函数内部做参数校验避免None传入。错误回传示例try: fn_args json.loads(tool_call.function.arguments or {}) except json.JSONDecodeError as e: messages.append({ role: tool, tool_call_id: tool_call.id, name: fn_name, content: json.dumps({error: f参数解析失败: {e}}), }) continue6.3 上下文无限膨胀导致成本上升现象任务执行到中后段模型响应变慢token 消耗快速上升甚至超出上下文限制。可能原因没有对 messages 做长度控制工具结果过大且每轮重复传递系统提示词写得过长且内容重复单次任务迭代次数没有上限。处理建议设置消息最大条数和最大 token 数对工具结果做摘要清理不再使用的中间信息超过阈值时提示用户“上下文过长请重新发起任务”。6.4 本地部署模型与云端 API 的行为差异现象本地部署的 DeepSeek 模型在同样代码下不触发工具调用或返回格式与云端不一致。可能原因本地推理服务没有启用 function calling 能力模型微调版本不同指令遵循能力有差异系统提示词不完全适用于本地模型API 协议兼容层不完整。处理建议先调用本地服务的/models接口确认协议版本使用官方提供的 OpenAI 兼容测试样例验证 function calling在 Harness 中做协议适配层允许切换不同后端对不同模型使用不同系统提示词和 schema 描述。7. 最佳实践与扩展方向7.1 从自研最小 Harness 到框架选型自研最小 Harness 适合学习和处理简单场景。当需求变复杂比如需要多 Agent 协作、长期记忆、复杂工作流、多人协作调试可以考虑引入成熟框架。选型时不要只看流行度要按以下能力评估能力为什么重要工具调用稳定性是否支持流式工具调用、错误恢复可观测性是否有内置 tracing、日志接口扩展性是否能自定义工具注册、消息处理社区活跃度遇到问题时是否能快速找到方案学习成本团队是否容易上手和维护7.2 可复用检查清单上线一个 Agent Harness 前至少检查以下项目[ ] API Key 是否通过环境变量注入不写死在代码和仓库中[ ] base_url、model 名称是否按环境拆分[ ] 工具 schema 与函数签名是否完全一致[ ] 工具函数是否做了参数校验和异常捕获[ ] 模型调用是否设置了超时和重试[ ] Agent 主循环是否设置了最大迭代次数[ ] 是否记录每次模型请求和工具执行的日志[ ] 是否对工具结果大小做了限制或摘要[ ] 是否对用户并发请求做了限流[ ] 是否对上线的模型版本和工具版本做了记录。这些清单可以直接作为团队 Code Review 的参考项。7.3 下一步学习路径如果想继续深入可以按这个路线扩展先跑通本文的最小 Harness加入两个不同类型的工具比如查询接口和更新数据。加上消息长度控制和异步日志观察多轮调用时上下文变化。接入真实业务数据库或内部 HTTP 服务替换掉示例中的硬编码数据。在 Harness 外部加一层 API 服务把用户请求、会话管理、鉴权、限流串起来。对比几个主流 Agent 框架的源码理解别人如何解决上下文管理、工具调用恢复、并发调度等问题。最后给一个直接的练习建议不要急着引入重型框架先用 DeepSeek API 和几十行 Python 搭一个最小 Harness跑通一次完整的工具调用。把“模型决策”和“工具执行”之间的消息流转理解清楚之后再去看企业级框架会发现很多设计都是围绕这些基础问题展开的。