DeepSeek Harness 完全指南：从零搭建你的第一个 AI Agent 工作流
目录
- 1. 什么是 AI Agent 工作流
- 2. 环境准备
- 3. Harness 的核心设计
- 4. 从零实现 Agent Harness
- 5. 编写你的第一个工具
- 6. 组装完整示例
- 7. 进阶：为 Harness 增加记忆能力
- 8. 常见问题与调试技巧
- 9. 进阶扩展方向
- 10. 总结
1. 什么是 AI Agent 工作流
在传统聊天机器人中，用户提问后模型只负责“生成一段回答”，任务到此结束。而 AI Agent 工作流则把大模型当作一个调度中枢：模型可以自主决定调用哪些工具、按什么顺序执行、如何根据中间结果调整下一步计划，最终完成一个多步骤任务。
一个典型的 Agent 工作流包含以下能力：
- 规划：把复杂目标拆解成可执行步骤。
- 工具调用：调用搜索、计算器、API、数据库等外部能力。
- 记忆：在多轮交互中保留上下文。
- 反思与重试：根据工具返回结果修正行动。
DeepSeek 提供的对话与推理模型具备较强的指令跟随和函数调用能力，非常适合作为 Agent 的“大脑”。本文会带你从零搭建一个轻量的 DeepSeek Harness，它负责统一管理模型调用、工具注册、循环执行和结果汇总，让你能快速构建自己的第一个 Agent 工作流。
2. 环境准备
开始之前，请先准备以下内容。
2.1 获取 DeepSeek API Key
- 打开 DeepSeek 开放平台并注册账号。
- 进入「API Keys」页面，点击「创建 API Key」。
- 复制生成的 key，妥善保存，不要提交到代码仓库。
2.2 安装依赖
DeepSeek 提供与 OpenAI 兼容的接口，因此可以直接使用 openai SDK，也可以使用原生 HTTP 请求。本文使用 openai SDK，因为它生态成熟、代码简洁。
pip install openai python-dotenv
建议把 key 放入 .env 文件：
DEEPSEEK_API_KEY=sk-xxxxxxxxxxxxxxxx
DEEPSEEK_BASE_URL=https://api.deepseek.com
DEEPSEEK_MODEL=deepseek-chat
2.3 项目结构
我们创建一个清晰的目录结构，后续代码都放在这里：
deepseek-harness/
├── .env
├── main.py
└── agent/
├── __init__.py
├── harness.py
└── tools.py
3. Harness 的核心设计
在动手写代码前，我们先明确 Harness 需要解决哪些问题：
- 统一模型接入：封装 API Key、base_url 和模型名称。
- 工具注册与描述：让模型知道有哪些工具可用，以及每个工具的参数格式。
- 多轮循环：模型可能多次调用工具，Harness 需要循环执行，直到模型给出最终答案。
- 状态管理：把每一轮的消息和工具结果都记录下来。
整体流程如下图所示：
4. 从零实现 Agent Harness
下面逐步完成 agent/harness.py。
4.1 初始化客户端
import os
from openai import OpenAI
from dotenv import load_dotenv
load_dotenv()
class DeepSeekHarness:
def __init__(self):
api_key = os.getenv("DEEPSEEK_API_KEY")
base_url = os.getenv("DEEPSEEK_BASE_URL")
self.model = os.getenv("DEEPSEEK_MODEL", "deepseek-chat")
if not api_key:
raise ValueError("缺少 DEEPSEEK_API_KEY，请检查 .env 文件")
self.client = OpenAI(api_key=api_key, base_url=base_url)
self.tools = []
4.2 注册工具
函数调用是 Agent 的核心。为了让 DeepSeek 识别工具，需要为每个工具提供 JSON Schema 描述。这里我们用一个装饰器来简化注册过程。
class Tool:
def __init__(self, name, description, parameters, handler):
self.name = name
self.description = description
self.parameters = parameters
self.handler = handler
def to_spec(self):
return {
"type": "function",
"function": {
"name": self.name,
"description": self.description,
"parameters": self.parameters,
},
}
接下来在 Harness 中增加注册方法：
def register_tool(self, name, description, parameters):
def decorator(func):
tool = Tool(name, description, parameters, func)
self.tools.append(tool)
return func
return decorator
4.3 组装提示词并执行循环
Agent 循环的关键是：调用模型后，如果返回 tool_calls，就逐个执行工具，并把结果追加到消息历史，然后再次调用模型，直到不再请求工具。
def run(self, user_input, system_prompt=None, max_steps=5):
messages = []
if system_prompt:
messages.append({"role": "system", "content": system_prompt})
messages.append({"role": "user", "content": user_input})
steps = 0
while steps < max_steps:
response = self.client.chat.completions.create(
model=self.model,
messages=messages,
tools=[t.to_spec() for t in self.tools] or None,
)
message = response.choices[0].message
if not message.tool_calls:
return message.content
messages.append(message)
for tool_call in message.tool_calls:
tool_name = tool_call.function.name
arguments = tool_call.function.arguments
result = self._execute_tool(tool_name, arguments)
messages.append({
"role": "tool",
"tool_call_id": tool_call.id,
"content": result,
})
steps += 1
return "已达到最大执行步数，请简化任务后重试。"
def _execute_tool(self, name, arguments):
import json
args = json.loads(arguments or "{}")
for tool in self.tools:
if tool.name == name:
try:
result = tool.handler(**args)
return str(result)
except Exception as exc:
return f"工具执行失败: {exc}"
return f"未找到工具: {name}"
5. 编写你的第一个工具
现在我们来创建几个实用的工具。在 agent/tools.py 中定义工具函数，并编写好参数 Schema。
5.1 计算器工具
def calculator_schema():
return {
"type": "object",
"properties": {
"expression": {
"type": "string",
"description": "需要计算的数学表达式，例如 '3 * (4 + 5)'",
}
},
"required": ["expression"],
}
5.2 获取当前时间工具
from datetime import datetime
def get_current_time():
return datetime.now().strftime("%Y-%m-%d %H:%M:%S")
5.3 模拟天气查询工具
为了演示方便，这里返回固定数据。实际项目中你可以替换为真实的天气 API。
def fetch_weather(city: str):
mock_data = {
"北京": "晴，28 摄氏度",
"上海": "小雨，24 摄氏度",
"深圳": "多云，30 摄氏度",
}
return mock_data.get(city, f"暂未收录 {city} 的天气数据")
6. 组装完整示例
回到 main.py，把前面实现的所有组件串起来：
from agent.harness import DeepSeekHarness
from agent.tools import calculator_schema, get_current_time, fetch_weather
def main():
agent = DeepSeekHarness()
agent.register_tool(
name="calculator",
description="执行数学计算",
parameters=calculator_schema(),
)(lambda expression: eval(expression, {"__builtins__": {}}, {}))
agent.register_tool(
name="get_current_time",
description="获取当前日期和时间",
parameters={"type": "object", "properties": {}},
)(get_current_time)
agent.register_tool(
name="fetch_weather",
description="查询指定城市的天气",
parameters={
"type": "object",
"properties": {
"city": {"type": "string", "description": "城市名称"}
},
"required": ["city"],
},
)(fetch_weather)
system_prompt = (
"你是一个乐于助人的 AI 助手。"
"当需要计算、查询时间或天气时，请使用提供的工具。"
"如果工具返回错误，请尝试修正参数后重试。"
)
question = "今天北京天气怎么样？顺便帮我算一下 25 * 4 + 10 等于多少。"
answer = agent.run(question, system_prompt=system_prompt)
print("Agent 回答：", answer)
if __name__ == "__main__":
main()
运行效果类似：
Agent 回答：今天北京晴朗，气温 28 摄氏度。25 * 4 + 10 的计算结果是 110。
7. 进阶：为 Harness 增加记忆能力
上面的 Agent 每次执行都会重置消息历史。如果希望支持多轮对话，可以把消息记录保存在实例中，并允许用户传入会话 ID。这样 Agent 就能记住之前聊过什么。
class DeepSeekHarness:
def __init__(self):
...
self.history = {}
def run(self, user_input, session_id="default", system_prompt=None, max_steps=5):
if session_id not in self.history:
self.history[session_id] = []
messages = self.history[session_id]
if system_prompt and not any(
m.get("role") == "system" for m in messages
):
messages.insert(0, {"role": "system", "content": system_prompt})
messages.append({"role": "user", "content": user_input})
steps = 0
while steps < max_steps:
response = self.client.chat.completions.create(
model=self.model,
messages=messages,
tools=[t.to_spec() for t in self.tools] or None,
)
message = response.choices[0].message
if not message.tool_calls:
messages.append(message)
return message.content
messages.append(message)
for tool_call in message.tool_calls:
result = self._execute_tool(
tool_call.function.name,
tool_call.function.arguments,
)
messages.append({
"role": "tool",
"tool_call_id": tool_call.id,
"content": result,
})
steps += 1
return "已达到最大执行步数，请简化任务后重试。"
使用会话 ID 后，连续提问可以共享上下文：
agent.run("上海天气如何？", session_id="user-1")
agent.run("那北京呢？", session_id="user-1")
8. 常见问题与调试技巧
8.1 模型没有调用工具
检查工具的 description 是否清晰，参数 Schema 是否完整，以及 tools 参数是否正确传递。同时确认使用的模型支持 function calling，deepseek-chat 是支持的。
8.2 工具参数解析失败
常见原因是工具返回了非 JSON 字符串，或模型生成的参数不符合 Schema。可以在 _execute_tool 中增加日志，打印原始参数：
import logging
logging.basicConfig(level=logging.INFO)
def _execute_tool(self, name, arguments):
logging.info("调用工具 %s，参数 %s", name, arguments)
...
8.3 循环达到最大步数
说明模型在多轮调用后仍未结束任务。可以适当调大 max_steps，但更建议检查工具描述是否让模型产生了重复调用。必要时在系统提示词中加入“如果多次调用同一工具仍失败，请直接告知用户无法完成”。
8.4 温度与输出稳定性
Agent 工作流通常希望输出更稳定，可以在创建客户端时统一设置较低温度：
self.client = OpenAI(
api_key=api_key,
base_url=base_url,
temperature=0.2,
)
注意：不同模型对参数支持情况可能不同，使用前请以 DeepSeek 官方文档为准。
9. 进阶扩展方向
当你的第一个 Agent 跑通之后，可以继续探索以下方向：
- 引入 LangChain 或 LlamaIndex，获得更丰富的工具生态。
- 加入向量数据库，实现 RAG 检索增强。
- 使用 Pydantic 校验工具参数，减少格式错误。
- 加入流式输出，提升交互体验。
- 增加并发控制与限流，避免 API 调用过载。
- 引入规划器，让 Agent 先生成步骤计划，再逐步执行。
这些能力都可以在现有 Harness 的基础上逐步叠加，而不需要推翻重写。
10. 总结
本文从零实现了一个轻量的 DeepSeek Harness，完成了以下关键步骤：
- 配置 DeepSeek API 环境。
- 设计工具注册机制。
- 实现多轮工具调用循环。
- 编写计算器、时间、天气三个示例工具。
- 增加会话记忆能力。
通过这个 Harness，你已经掌握了构建 AI Agent 工作流的基本骨架。下一步可以把你工作中的真实 API 包装成工具接入进来，让 Agent 真正解决实际问题。随着工具越来越丰富，Agent 的自主能力也会越来越强，最终形成一套可复用的智能工作流。
DAMO开发者矩阵，由阿里巴巴达摩院和中国互联网协会联合发起，致力于探讨最前沿的技术趋势与应用成果，搭建高质量的交流与分享平台，推动技术创新与产业应用链接，围绕“人工智能与新型计算”构建开放共享的开发者生态。
更多推荐
0
0- 0
已为社区贡献1条内容
所有评论(0)