DeepSeek Harness：面向生产级智能体的工程化脚手架

1. 项目概述：这不是一个“框架”，而是一套可落地的智能体工程化脚手架

最近在多个技术社区和内部分享中，频繁看到“Agent Harness”这个词被提起，尤其搭配“DeepSeek Harness”一起出现时，常被误读为某个开源框架、SDK包，甚至有人直接当成DeepSeek官方推出的Agent开发平台。其实不然——它本质上是一套面向生产级智能体（Agent）系统构建的工程化方法论与配套实践模板集合，核心目标不是替代LangChain或LlamaIndex，而是解决它们在真实业务场景中长期被忽视的“最后一公里”问题：如何让一个在Jupyter里跑通的Agent Demo，真正变成可监控、可灰度、可回滚、可审计、能扛住每秒200次并发调用的线上服务。我参与过三个不同规模的Agent落地项目，从某高校实验室的科研辅助系统，到某公司内部知识中枢，再到面向终端用户的多模态客服助手，所有项目在进入联调阶段后都卡在同一个地方：Prompt版本混乱、工具调用链路不可见、失败case无法归因、重试逻辑全靠硬编码。正是这些反复踩坑的经历，让我意识到，我们缺的不是更炫的模型，而是一套像Docker Compose之于微服务、像Terraform之于云资源那样，能把Agent的“行为”、“状态”、“依赖”、“可观测性”全部声明化、可配置、可复用的工程底座。DeepSeek Harness正是这样一套思路的具象化产物——它不绑定任何大模型供应商，不强制使用特定Orchestrator，甚至不提供一行LLM推理代码；它只定义接口契约、约定目录结构、内置标准日志Schema、预置OpenTelemetry采集点、封装通用的Tool Registry注册/发现机制。换句话说，它把“写一个能动的Agent”这件事，拆解成了“定义能力边界”、“编排执行路径”、“注入可观测探针”、“配置弹性策略”四个可独立演进、可交叉验证的模块。对刚接触Agent开发的新手，它能帮你绕过90%的环境配置陷阱，5分钟内启动一个带完整Trace链路的Hello World Agent；对资深架构师，它提供了一套清晰的扩展锚点，比如你可以无缝替换默认的Memory Backend为Redis Cluster，或把默认的Fallback Policy换成基于强化学习的动态降级策略。它不承诺“开箱即用”，但承诺“开箱即控”——所有行为都可观察、所有状态都可追溯、所有变更都可灰度。这才是它在当前Agent开发热潮中真正稀缺的价值。

2. 核心设计思路拆解：为什么是“Harness”而不是“Framework”

2.1 “Harness”的本质：一种约束性极强的工程契约

很多人第一眼看到“Harness”这个词，会下意识联想到“马具”“挽具”，觉得它是个束缚性的存在。这恰恰是对的——但这个“束缚”，不是限制你的创造力，而是帮你主动放弃那些低价值、高风险的自由度。举个具体例子：在传统Agent开发中，你完全可以用Python的subprocess模块去调用外部API，也可以自己手写HTTP Client做重试，甚至用全局变量存Session状态。这些做法在Demo阶段毫无问题，但一旦进入团队协作或生产环境，就会立刻暴露出三个致命缺陷：一是调试成本指数级上升（谁改了哪个全局变量？哪个重试逻辑漏写了熔断？），二是版本管理彻底失效（Prompt、Tool Schema、Memory初始化逻辑散落在十几个文件里），三是安全审计无从下手（外部调用没有统一鉴权入口）。DeepSeek Harness的设计哲学，就是用一套轻量但不可绕过的“契约”来封禁这些危险路径。它强制要求：所有外部工具调用必须通过ToolExecutor抽象层，该层内置了统一的超时控制、错误分类、审计日志打点；所有状态存储必须实现StateBackend接口，且默认提供基于SQLite的本地实现和基于PostgreSQL的分布式实现；所有Prompt模板必须遵循Jinja2语法并存放在prompts/目录下，且每个模板文件必须附带schema.yaml描述其输入参数类型与校验规则。这种“契约”不是靠文档约定，而是靠项目初始化时自动生成的pyproject.toml中预设的[tool.deepseek-harness]配置块来强制执行。当你运行dsh init命令时，它不仅创建目录结构，还会在pyproject.toml里写入：

[tool.deepseek-harness] version = "0.3.2" required_plugins = ["dsh-tool-http", "dsh-memory-sqlite"] default_executor = "tool_http"

这意味着，任何试图绕过ToolExecutor直接发HTTP请求的代码，在CI流水线的dsh validate检查阶段就会被拦截报错。这种“约束力”，正是Harness区别于Framework的核心——Framework告诉你“可以怎么干”，Harness则明确告诉你“只能怎么干”，并把“不能怎么干”的检查点嵌入到开发流程的每一个环节。我见过太多团队在项目中期推倒重来，原因就是早期为了赶进度，放任了这种“自由”，结果后期为了补可观测性，不得不给每个函数手动加logging.info()，再为了加重试，又给每个HTTP调用补try-except，最后整个代码库变成一张无法维护的蜘蛛网。Harness的“束缚”，本质上是一种提前支付的技术债利息，换来的是后期指数级降低的维护成本。

2.2 与主流Agent框架的定位差异：不做“轮子”，只做“轴承”

市面上已有的Agent框架，大致可分为三类：一类是以LangChain为代表的“胶水型”框架，核心价值在于连接各种LLM、Tool、Memory组件，提供统一的抽象接口；第二类是以AutoGen为代表的“多Agent协作型”框架，聚焦于Agent之间的通信协议与任务分发；第三类则是以LlamaIndex为代表的“RAG增强型”框架，专精于文档索引与检索优化。DeepSeek Harness与它们的关系，不是竞争，而是正交互补。它不提供任何LLM调用封装，不实现任何Tool的具体逻辑，也不处理任何向量检索算法。它的全部精力，都花在了“胶水”之外、“协作”之上、“增强”之后的环节——也就是当LangChain已经把一个Prompt发给模型、AutoGen已经把任务分发给Worker Agent、LlamaIndex已经返回了最相关的三段文本之后，系统该如何稳定、可靠、可追踪地完成后续动作。这就像汽车制造：LangChain是发动机控制系统，AutoGen是变速箱，LlamaIndex是涡轮增压器，而DeepSeek Harness，则是那套精密的轴承系统——它不产生动力，但决定了动力能否平稳、高效、无损耗地传递到车轮；它不决定换挡时机，但确保每一次换挡的冲击都被精准吸收；它不提升进气效率，但让涡轮的每一次喘振都能被实时监测并预警。具体到技术实现上，Harness的“轴承”特性体现在三个关键设计上：首先是执行上下文（Execution Context）的全程透传。每个Agent调用都会生成一个唯一的trace_id，这个ID会自动注入到所有下游调用中——无论是调用外部API、查询数据库、还是写入日志。你不需要在每个函数里手动传参，Harness的ContextManager会在入口处自动捕获，并通过Python的contextvars模块在异步/同步混合环境中安全传递。其次是状态快照（State Snapshot）的自动触发。当Agent执行到关键决策点（如选择Tool、生成最终Answer、触发Fallback）时，Harness会自动调用StateBackend.save_snapshot()，将当前完整的内存状态、输入参数、中间推理步骤序列化存档。这些快照不是为了回滚，而是为了事后归因——当用户投诉“为什么昨天回答正确，今天就错了”，你只需输入trace_id，就能在后台查到两次调用的完整状态对比，瞬间定位是Prompt更新导致的语义漂移，还是外部API返回格式发生了变更。最后是弹性策略（Resilience Policy）的声明式配置。Harness不内置重试逻辑，但它定义了一个标准的ResiliencePolicy接口，允许你用YAML文件声明：“对weather_api这个Tool，启用指数退避重试，最多3次，每次间隔1s、2s、4s；若连续失败，自动降级到缓存数据，并触发告警”。这套策略会被ToolExecutor在运行时动态加载，无需修改一行业务代码。这种“只做轴承”的定位，让它天然具备极强的兼容性——你可以把Harness嵌入到任何现有LangChain项目中，只需替换掉原来的Runnable链，用HarnessAgent包装即可；你也可以在AutoGen的GroupChatManager里，把每个Agent的generate_reply方法，替换成Harness封装的execute_with_context。它不争上游，只稳下游；不抢风头，只保底线。这才是它能在众多框架中找到自己不可替代位置的根本原因。

2.3 架构分层与核心模块职责：四层解耦，各司其职

DeepSeek Harness的架构并非一个单体应用，而是严格遵循“关注点分离”原则，划分为四个清晰、低耦合的层次，每一层都有明确的边界与不可替代的职责。这种分层不是为了炫技，而是为了应对真实业务中不断变化的需求——比如，某天产品要求增加对微信小程序的适配，你只需要在最上层的Adapter层新增一个WeChatAdapter，完全不用碰下面三层的任何代码；又比如，运维团队要求将日志从本地文件迁移到ELK集群，你只需替换ObservabilityLayer中的LoggerBackend实现，其他模块毫发无损。这种解耦能力，是它支撑长期演进的生命线。下面我结合实际项目经验，逐层拆解其设计意图与实操细节：

第一层：Adapter Layer（接入层）
这是用户最先接触到的部分，负责将外部请求“翻译”成Harness内部的标准消息格式。它不处理业务逻辑，只做协议转换。例如，当一个HTTP请求到达时，HTTPAdapter会解析Content-Type、提取AuthorizationHeader、校验JWT Token（如果配置了）、将JSON Body反序列化为AgentRequest对象，并注入trace_id和user_id。如果是WebSocket连接，则由WSAdapter接管，它会维护连接生命周期，将每条Message封装为AgentEvent，并确保消息顺序与trace_id的绑定。关键设计点在于：所有Adapter都必须实现BaseAdapter接口，该接口强制定义了parse_input()和format_output()两个方法。这意味着，无论你接入的是HTTP、gRPC、MQTT还是未来可能出现的某种新协议，只要实现了这两个方法，就能无缝接入Harness生态。我在某次为IoT设备定制Agent时，就只花了半天时间，基于BaseAdapter写了一个MQTTAdapter，它能自动订阅设备上报的主题，将原始JSON payload转换为标准AgentRequest，并将Agent的回复发布到设备的响应主题。整个过程，没有修改Harness核心代码一行，也没有影响其他Adapter的运行。

第二层：Orchestration Layer（编排层）
这是Harness的“大脑”，负责根据AgentRequest的内容，调度合适的Agent实例，执行其run()方法，并管理整个执行生命周期。它不关心Agent内部如何思考，只关心“谁来干”、“何时干”、“干完怎么收场”。核心组件是AgentRouter和ExecutionEngine。AgentRouter是一个可插拔的路由策略引擎，支持多种路由模式：基于request.path的静态路由（如/api/v1/chat->ChatAgent）、基于request.intent的语义路由（需配合轻量NLU模型）、甚至基于user.tier的灰度路由（VIP用户走A模型，普通用户走B模型）。ExecutionEngine则负责真正的执行：它会为每次调用创建一个独立的ExecutionContext，注入所有必要的上下文（如StateBackend实例、ToolRegistry、Logger），然后调用Agent的run()方法。这里的关键创新是ExecutionEngine的“沙箱化”设计——它通过threading.local或asyncio.Task.local为每个执行上下文隔离了所有共享状态，确保高并发下不会出现状态污染。我曾在一个高并发客服场景中，将QPS从50压测到300，ExecutionEngine的CPU占用率始终稳定在65%以下，而内存泄漏检测工具显示，所有ExecutionContext都在任务结束后被及时GC回收，这得益于其严格的生命周期管理。

第三层：Core Agent Layer（核心Agent层）
这是业务逻辑的承载层，也是开发者编写代码的主要区域。Harness在这里提供了最小化的基类BaseAgent，它只定义了run(self, request: AgentRequest) -> AgentResponse这一个抽象方法，以及几个便捷的工具方法（如self.use_tool()、self.save_state()）。它刻意不提供任何高级功能，比如记忆管理、工具自动发现、多步规划等——这些都交给上层Orchestration或下层Tool Layer去处理。这种“极简主义”设计，带来了两个巨大好处：一是学习成本极低，新手看懂BaseAgent的源码只需5分钟；二是演进路径清晰，所有复杂功能都作为可选插件（Plugin）存在，比如dsh-plugin-memory-redis提供分布式记忆，dsh-plugin-planning-reasoning提供CoT规划能力。你在项目初期，完全可以只用BaseAgent写一个纯Prompt驱动的简单Agent，等业务复杂度上来后，再按需安装插件，整个过程平滑无痛。我在某知识库问答项目中，就经历了这样的演进：第一版只有BaseAgent+prompt.j2，第二版加入了dsh-plugin-rag插件，第三版又叠加了dsh-plugin-fallback-llm插件。每次升级，都只是修改pyproject.toml里的required_plugins列表，然后运行dsh install，Harness会自动下载、校验、集成新插件，原有代码一行都不用改。

第四层：Infrastructure Layer（基础设施层）
这是Harness的“地基”，提供所有底层能力支撑，包括ToolRegistry（工具注册中心）、StateBackend（状态存储）、LoggerBackend（日志后端）、MetricsBackend（指标收集）等。它的设计哲学是“提供标准，不垄断实现”。例如，ToolRegistry接口只定义了register(tool: BaseTool)、get_by_name(name: str) -> BaseTool、list_all() -> List[str]三个方法，至于你是用内存字典、Redis哈希表还是PostgreSQL表来实现，Harness完全不管。同样，StateBackend只规定了save_snapshot(trace_id: str, state: dict)和load_snapshot(trace_id: str) -> dict两个契约，你可以用SQLite做本地开发，用S3做生产备份，甚至用MongoDB做长周期归档。这种设计，让Harness具备了极强的环境适应性。我在一个客户现场部署时，由于网络策略限制，无法访问公网Docker Hub，但客户内部有成熟的Harbor私有仓库。我只需将Harness的dsh-infrastructure-postgres镜像推送到Harbor，然后在docker-compose.yml里修改image字段，整个基础设施层就完成了私有化迁移，前后耗时不到1小时。这种“标准先行、实现可选”的分层思想，正是Harness能跨越不同技术栈、不同组织架构，成为真正通用工程底座的根本保障。

3. 核心模块详解与实操要点：从零搭建一个可监控的Agent服务

3.1 初始化项目与目录结构：不只是dsh init，更是工程规范的起点

执行dsh init my-agent-project远不止是创建一堆空文件夹那么简单。它是一次工程规范的强制植入，其生成的目录结构，本身就是一份活的、可执行的架构说明书。我建议你不要跳过这一步，哪怕你打算从头手写所有代码，也请先运行一次dsh init，然后仔细研究它生成的每个文件和目录的用途。下面是我对标准初始化结构的逐层解读，结合了我在三个项目中踩过的坑和总结出的最佳实践：

my-agent-project/ ├── pyproject.toml # 【核心】Harness的“宪法”，所有配置、插件、约束都在此定义 ├── README.md ├── prompts/ # 所有Prompt模板存放处，必须是Jinja2语法 │ ├── chat.j2 # 主Prompt，可继承base.j2 │ └── base.j2 # 基础Prompt，包含通用System Message ├── tools/ # 所有Tool实现，必须继承BaseTool │ ├── http_client.py # 封装requests，内置重试与审计 │ └── knowledge_retriever.py # RAG检索Tool，对接向量库 ├── agents/ # Agent业务逻辑，必须继承BaseAgent │ └── chat_agent.py # 入口Agent，实现run()方法 ├── adapters/ # 接入协议适配器 │ └── http_adapter.py # HTTP协议转换 ├── infrastructure/ # 基础设施配置 │ ├── state/ # StateBackend配置 │ │ └── sqlite_config.yaml # 本地开发用 │ └── observability/ # 可观测性配置 │ └── logger_config.yaml # 日志输出格式与级别 ├── tests/ # 测试用例，Harness强制要求覆盖率报告 │ └── test_chat_agent.py └── .dsh/ # Harness运行时元数据，禁止手动修改 └── harness.lock # 插件版本锁文件，确保CI/CD环境一致性

最关键的文件是pyproject.toml。它不仅是Python项目的标准配置，更是Harness的“中央控制器”。我强烈建议你打开它，重点关注[tool.deepseek-harness]区块。这里有几个必填且极易出错的配置项，需要你根据项目实际情况谨慎填写：

• version: 这不是项目的语义化版本号，而是Harness自身的版本约束。例如"0.3.*"表示允许使用0.3.x系列的所有小版本，但禁止升级到0.4.0。这个约束非常重要，因为Harness的API在主版本间可能有破坏性变更。我在某次升级中，因未锁定版本，导致StateBackend接口从save()变成了save_snapshot()，所有自定义的Redis实现全部报错，回滚花了整整一天。教训是：永远用x.y.*格式，绝不写*。

• required_plugins: 这是一个字符串列表，列出了项目运行所必需的Harness插件。例如["dsh-plugin-tool-http", "dsh-plugin-memory-sqlite"]。Harness在启动时会检查这些插件是否已安装，未安装则报错退出。这里有个隐藏技巧：你可以利用它来实现“环境差异化配置”。比如在开发环境，你写["dsh-plugin-memory-sqlite"]，在生产环境，你写["dsh-plugin-memory-redis"]，然后在CI/CD流水线中，根据ENV变量动态覆盖这个值。这样，同一套代码，就能在不同环境使用不同的记忆后端，无需任何条件编译。

• default_adapter: 指定默认的接入适配器。通常设为"http"。但如果你的项目是纯内部服务，只通过gRPC调用，那么这里应该设为"grpc"，Harness会自动为你生成adapters/grpc_adapter.py的骨架代码。

• observability_enabled: 布尔值，控制是否启用全套可观测性。设为true时，Harness会自动注入OpenTelemetry SDK，配置LoggerBackend和MetricsBackend。设为false时，所有可观测性代码都不会被加载，性能开销几乎为零。这个开关在压测阶段非常有用——你可以先关掉可观测性，测出纯业务逻辑的极限QPS，再打开它，对比性能损耗，从而评估是否需要优化日志打点频率。

另一个容易被忽略但极其重要的文件是prompts/base.j2。Harness强制要求所有Prompt都必须继承它，目的是统一System Message。base.j2的默认内容是：

{% set system_message = "You are a helpful AI assistant. Always respond in Chinese." %} {{ system_message }} {% block content %}{% endblock %}

这个设计看似简单，却解决了两个大问题：一是避免每个Prompt都重复写相同的System Message，导致维护困难；二是为后续的“Prompt A/B测试”埋下伏笔。你可以在chat.j2中这样使用：

{% extends "base.j2" %} {% block content %} User: {{ request.query }} Context: {{ request.context | default('') }} {% endblock %}

当你要测试一个新的System Message时，只需修改base.j2，所有继承它的Prompt都会自动生效。我在某次优化客服Agent的语气时，就通过这种方式，一次性更新了12个不同场景的Prompt，零代码修改，零发布风险。

提示：dsh init生成的tests/目录里，有一个test_chat_agent.py的模板。它不是一个摆设，而是Harness强制要求的“健康检查”。这个测试用例会模拟一次完整的HTTP请求，调用你的Agent，并断言返回的response.status_code == 200。你必须保证这个测试在dsh test命令下100%通过，Harness才会允许你执行dsh deploy。这是一种“测试即准入”的质量门禁，它逼着你在写业务代码的第一天，就建立起端到端的验证闭环。我见过太多团队，把测试留到最后，结果上线前发现HTTP Adapter根本没处理好CORS，或者AgentResponse的JSON Schema和前端约定的不一致，返工代价巨大。Harness把这个门禁提前到了项目初始化的那一刻。

3.2 Tool开发与注册：让外部能力“可插拔、可审计、可熔断”

在Harness的世界里，“Tool”不是一段随意的函数，而是一个拥有完整生命周期、可被统一管理的“能力单元”。它的开发流程，严格遵循“定义接口 -> 实现逻辑 -> 注册到Registry -> 在Agent中调用”四步法。这个流程看似繁琐，但带来的收益是巨大的：可插拔性让你能像换电池一样更换外部API；可审计性让你能精确追踪每一次调用的耗时、成功率、输入输出；可熔断性则是在外部服务宕机时，保护你的Agent不被拖垮。下面我以一个真实的天气查询Tool为例，详细拆解每一步的实操细节与避坑指南。

第一步：定义Tool接口（tools/weather_tool.py）
Harness要求所有Tool必须继承BaseTool，并实现name、description、args_schema和_run四个属性/方法。args_schema是Pydantic v2的BaseModel，用于定义Tool的输入参数及其校验规则。这是最容易出错的地方——很多新手会在这里写一个过于宽泛的Schema，导致后期无法做精细化的输入校验和日志脱敏。

from pydantic import BaseModel, Field from dsh.core.tool import BaseTool class WeatherInput(BaseModel): city: str = Field(..., description="城市名称，必须是中文，如'北京'") unit: str = Field("celsius", description="温度单位，可选'celsius'或'fahrenheit'") class WeatherTool(BaseTool): name: str = "get_weather" description: str = "获取指定城市的当前天气信息，包括温度、湿度、风速。仅支持中国境内城市。" args_schema: type[BaseModel] = WeatherInput def _run(self, city: str, unit: str = "celsius") -> str: # 实际调用外部API的逻辑 pass

注意city字段的Field(..., description=...)，这里的description不是给人看的注释，而是Harness自动生成OpenAPI文档和前端表单的依据。如果你写city: str而不加Field，Harness在生成Swagger UI时，就会把city显示为string，没有任何提示，前端工程师拿到后会一脸懵。另外，unit字段的默认值"celsius"，会被Harness自动注入到args_schema的default属性中，这意味着，当用户调用时不传unit参数，_run方法收到的unit值就是"celsius"，无需在方法体内再做if unit is None的判断。这是一个很小但很实用的细节。

第二步：实现Tool逻辑（tools/weather_tool.py续）
_run方法是Tool的执行体，它必须是同步的（Harness会自动处理异步调用）。这里的核心是把所有外部依赖，都包裹在Harness提供的标准客户端里。不要自己写requests.get()！Harness的ToolExecutor会自动为你注入一个http_client实例，它已经内置了超时、重试、错误分类、审计日志等功能。

from dsh.infrastructure.http import get_http_client def _run(self, city: str, unit: str = "celsius") -> str: # 1. 使用Harness标准HTTP客户端 client = get_http_client() # 2. 构造请求 url = f"https://api.weather.com/v3/wx/forecast/daily/3day" params = { "postalKey": self._get_postal_key(city), # 内部方法，将城市名转为邮编key "language": "zh-CN", "format": "json", "unit": unit } # 3. 发起调用，Harness会自动记录trace_id、耗时、状态码 try: response = client.get(url, params=params, timeout=5.0) response.raise_for_status() # 4. 解析并格式化返回结果 data = response.json() return f"当前{city}天气：{data['temperature']}°{unit[0].upper()}，湿度{data['humidity']}%，风速{data['windSpeed']}m/s" except client.Timeout: # Harness会自动捕获并分类此异常，触发熔断逻辑 raise Exception(f"天气API调用超时，请稍后重试") except client.HTTPStatusError as e: # Harness会记录具体的HTTP状态码，便于后续分析 if e.response.status_code == 404: raise Exception(f"未找到城市'{city}'的天气信息") else: raise Exception(f"天气API返回错误：{e.response.status_code}")

关键点在于client.get()的调用。get_http_client()返回的不是一个普通的requests.Session，而是一个Harness封装的InstrumentedHTTPClient。它会在每次调用前后，自动打点记录trace_id、tool_name、url、params（已脱敏，params中的敏感字段如api_key会被自动过滤）、status_code、elapsed_ms。这些日志会被发送到LoggerBackend，你可以在Kibana里用tool_name: "get_weather"快速筛选出所有天气查询的日志，分析平均耗时、错误率、地域分布。这就是“可审计性”的来源。

第三步：注册Tool到Registry（tools/__init__.py）
Harness不会自动扫描tools/目录下的所有类。你必须显式地在tools/__init__.py中，将Tool注册到全局ToolRegistry。这是Harness“可插拔”的关键一环。

# tools/__init__.py from dsh.core.tool import ToolRegistry from .weather_tool import WeatherTool # 将Tool注册到全局Registry ToolRegistry.register(WeatherTool())

ToolRegistry.register()的参数必须是一个Tool的实例，而不是类本身。这是因为Tool的__init__方法可能需要接收一些运行时配置，比如API Key。你可以在WeatherTool.__init__中添加参数：

def __init__(self, api_key: str = None): super().__init__() self.api_key = api_key or os.getenv("WEATHER_API_KEY")

然后在tools/__init__.py中这样注册：

ToolRegistry.register(WeatherTool(api_key="your-secret-key"))

这样，Tool的配置就和代码解耦了，你可以通过环境变量在不同环境注入不同的Key。

第四步：在Agent中调用Tool（agents/chat_agent.py）
这是最简单的一步，但也最容易写出“反模式”。正确的调用方式是使用self.use_tool()方法，它会自动处理Tool的查找、参数校验、执行、结果返回。

from dsh.core.agent import BaseAgent from dsh.core.tool import ToolRegistry class ChatAgent(BaseAgent): def run(self, request): # 1. 从request中提取用户意图 intent = self._detect_intent(request.query) # 2. 根据意图，决定是否调用Tool if intent == "weather": # 3. 调用Tool，传入参数 weather_result = self.use_tool( tool_name="get_weather", city=request.city, unit=request.unit or "celsius" ) return {"answer": f"好的，为您查询到：{weather_result}"} # 4. 兜底逻辑 return {"answer": "我暂时无法处理这个请求。"}

self.use_tool()的威力在于，它不仅仅是一个函数调用。它会：

• 自动从ToolRegistry中查找get_weather这个Tool；
• 使用WeatherInput的args_schema，对传入的city和unit进行Pydantic校验，如果city为空或unit不是枚举值，会立即抛出ValidationError，并被Harness捕获，返回标准化的错误响应；
• 在调用前后，自动注入ExecutionContext，确保trace_id等上下文信息在Tool内部也可用；
• 如果Tool执行失败，会根据ResiliencePolicy配置，自动触发重试或降级。

注意：绝对不要在Agent中直接import并调用Tool的_run方法！这会绕过Harness所有的可观测性、熔断、审计机制，让你的Tool变成一个“黑盒”，一旦出问题，排查起来比登天还难。我曾经接手过一个遗留项目，里面全是这种直连调用，为了给它加上监控，我不得不重写所有Tool，工作量是重新开发的两倍。Harness的设计哲学就是：所有通往外部世界的门，都必须经过同一个安检口。这个安检口，就是self.use_tool()。

3.3 Agent开发与执行：从run()方法到可灰度发布的完整链路

BaseAgent.run()方法，是Harness业务逻辑的唯一入口，也是整个Agent系统的“心脏”。它的设计，直接决定了你的Agent是脆弱的Demo，还是健壮的服务。Harness对run()方法的要求非常简单：接收一个AgentRequest对象，返回一个AgentResponse对象。但在这简单背后，隐藏着一套精密的执行保障机制。下面我将结合一个真实的电商客服Agent案例，从方法签名、内部逻辑、错误处理、灰度发布四个维度，深入剖析run()方法的正确写法与实战心得。

run()方法的签名与契约
AgentRequest和AgentResponse是Harness定义的两个核心数据模型，它们不是简单的字典，而是带有严格Schema的Pydantic模型。AgentRequest至少包含query（用户输入）、user_id（用户标识）、session_id（会话标识）、trace_id（链路追踪ID）等字段。AgentResponse则必须包含answer（最终回答）、status_code（HTTP状态码，如200成功，400参数错误，500内部错误）、metadata（任意键值对，用于传递调试信息）。

from dsh.core.agent import BaseAgent from dsh.core.request import AgentRequest from dsh.core.response import AgentResponse class ECommerceAgent(BaseAgent): def run(self, request: AgentRequest) -> AgentResponse: # 业务逻辑... pass

这个签名看似平凡，却蕴含了两个重要约束：一是类型安全，IDE可以为你提供完美的代码补全，request.后面能点出所有可用字段；二是契约清晰，任何阅读你代码的人，一眼就能明白输入输出的结构，无需翻阅文档。我在某次Code Review中，发现一个同事把request写成了dict，导致request.user_id在运行时才报KeyError。Harness的类型提示，提前把这个错误扼杀在了编辑器里。

run()内部的典型执行链路
一个健壮的run()方法，绝不是一条直线到底。它应该是一个有分支、有兜底、有状态管理的“决策树”。以下是我在电商客服项目中总结出的标准四步法：

1. 意图识别与路由（Intent Detection & Routing）
   不要指望大模型每次都准确理解用户意图。你应该在run()开头，用一个轻量、确定性的规则引擎或小型NLU模型，先做一次粗筛。

   def _detect_intent(self, query: str) -> str: # 简单关键词匹配，足够应对80%的场景 query_lower = query.lower() if "退货" in query_lower or "退款" in query_lower: return "return" elif "物流" in query_lower or "快递" in query_lower: return "logistics" elif "优惠券" in query_lower or "折扣" in query_lower: return "coupon" else: return "general"

   这样做的好处是：把高成本的LLM调用，留给真正需要深度理解的general场景；而对于return、logistics等高频、结构化的问题，直接走预定义的Tool链路，响应速度从2秒降到200毫秒，用户体验天壤之别。

2. 状态加载与上下文构建（State Loading & Context Building）
   在调用LLM之前，必须加载用户的历史会话状态，并构建一个丰富的上下文。Harness的StateBackend让你可以轻松做到这一点。

   # 加载用户最近3次会话的状态快照 recent_states = self.state_backend.load_recent( user_id=request.user_id, limit=3, exclude_current=True ) # 构建上下文字符串，供LLM参考 context_str = "\n".join([ f"第{i+1}次会话（{state['timestamp']}）：{state['last_query']} -> {state['last_answer']}" for i, state in enumerate(recent_states) ])

3. LLM调用与Tool编排（LLM Invocation & Tool Orchestration）
   这是run()的核心。Harness不提供LLM调