最近在探索大语言模型应用开发时发现一个现象很多开发者尝试将 DeepSeek 模型接入到各种 Agent 框架中但常常在环境配置、工具调用和流程编排上遇到瓶颈。与此同时一个名为 “Harness” 的概念在技术社区中频繁出现它似乎为 DeepSeek 的 Agent 化应用提供了一套更系统、更工程化的解决方案。本文将深入探讨如何基于 DeepSeek 模型构建一个功能完备的 AI Agent并重点解析 “Harness” 工程化实践的核心思想、技术实现与最佳路径。无论你是想快速搭建一个智能对话助手还是希望构建一个能处理复杂任务的企业级智能体本文提供的从零到一的完整指南和避坑方案都能为你提供直接可复用的参考。1. 背景与核心概念从 DeepSeek 到智能体工程在深入实战之前我们需要厘清几个关键概念这有助于理解整个技术栈的全貌。1.1 DeepSeek 模型强大的基座能力DeepSeek 是由深度求索公司开发的一系列大型语言模型。它因其在代码生成、逻辑推理和中文理解方面的出色表现而受到开发者社区的广泛关注。与一些通用聊天模型不同DeepSeek 的多个版本如 DeepSeek-Coder在编程任务上进行了深度优化使其成为构建开发辅助、自动化脚本生成等 Agent 的理想“大脑”。我们可以通过其提供的 API 来调用模型能力这是构建 Agent 的起点。1.2 AI Agent从“聊天”到“做事”AI Agent智能体不同于简单的聊天机器人。一个真正的 Agent 应具备以下核心能力感知与理解解析用户的自然语言指令理解其深层意图。规划与决策将复杂任务分解为可执行的子步骤序列。工具使用调用外部工具如搜索引擎、数据库、代码执行环境来获取信息或执行操作。记忆与学习在对话或任务执行过程中保持上下文并能从历史中学习。简单来说一个调用 DeepSeek API 的对话程序只是一个“问答机”而一个集成了工具调用、具备任务规划能力的 DeepSeek Agent 则是一个可以自主“做事”的智能助手。1.3 HarnessAgent 的工程化“缰绳”“Harness”在工程领域常指“线束”或“控制装置”。在 AI Agent 的语境下Harness 指的是一套用于控制、编排、测试和保障 Agent 稳定可靠运行的工程化框架和最佳实践集合。它解决了 Agent 开发中的常见痛点流程失控Agent 的思维链可能发散需要约束其行为边界。工具混乱多个工具如何被安全、高效地调用和管理。状态管理复杂的多轮对话和任务执行状态如何持久化和恢复。测试与评估如何系统化地测试 Agent 在各种场景下的表现。部署与监控如何将 Agent 部署到生产环境并监控其运行状态。你可以将 Harness 理解为 Agent 开发中的“Spring Framework”它提供了构建生产级智能体所需的基础设施和设计模式。网络上讨论的 “Harness Engineering” 或 “Agent Harness” 正是聚焦于这方面的工程实践。2. 环境准备与版本说明在开始构建我们的 DeepSeek Agent 之前需要准备好开发环境。以下配置是一个通用性较强的起点你可以根据自己的系统进行调整。核心环境要求操作系统Windows 10/11, macOS 10.15或主流 Linux 发行版如 Ubuntu 20.04。本文示例基于 Ubuntu 22.04。Python版本 3.8 - 3.11。推荐使用 3.9 或 3.10 以获得最佳的库兼容性。使用python --version检查。包管理工具pip通常随 Python 安装。DeepSeek API 密钥你需要访问 DeepSeek 平台并注册获取 API Key这是调用模型服务的凭证。代码编辑器VS Code、PyCharm 等任选。推荐 VS Code并安装 Python 扩展。项目初始化首先创建一个干净的项目目录并初始化虚拟环境这是管理 Python 项目依赖的最佳实践。# 创建项目目录 mkdir deepseek-agent-harness cd deepseek-agent-harness # 创建虚拟环境Python 3.9示例 python3.9 -m venv venv # 激活虚拟环境 # Linux/macOS source venv/bin/activate # Windows # venv\Scripts\activate # 升级pip pip install --upgrade pip3. 核心依赖与 Harness 框架选型构建一个具备 Harness 工程化特性的 Agent我们需要选择合适的库。我们将以LangChain和LangGraph作为核心框架因为它们提供了强大的 Agent 抽象和流程编排能力并且社区活跃生态丰富。安装核心依赖# 安装LangChain核心及社区工具包 pip install langchain langchain-community # 安装LangGraph用于构建有状态的、循环的Agent工作流 pip install langgraph # 安装DeepSeek的LangChain集成包如果官方提供或通用的OpenAI兼容层 # 由于DeepSeek API可能与OpenAI格式兼容我们通常使用openai库但需配置自定义base_url pip install openai # 安装用于网页搜索的工具依赖示例工具 pip install duckduckgo-search # 安装环境变量管理库 pip install python-dotenv版本说明与兼容性提示langchain和langgraph版本迭代较快本文示例基于langchain0.1.0和langgraph0.0.20的较新版本。如果遇到语法错误请查阅对应版本的官方文档。DeepSeek 的 API 端点可能更新请以官方最新文档为准。虚拟环境能有效隔离依赖避免与系统其他Python项目冲突务必在激活状态下进行后续操作。4. 构建基础 DeepSeek Agent让我们从构建一个最简单的、能调用 DeepSeek 模型并回答问题的 Agent 开始。4.1 配置模型访问首先在项目根目录创建.env文件来安全地存储你的 API 密钥切勿将密钥硬编码在代码中。# .env 文件 DEEPSEEK_API_KEYyour_deepseek_api_key_here DEEPSEEK_API_BASEhttps://api.deepseek.com/v1 # 请根据官方文档确认最新端点 DEEPSEEK_MODELdeepseek-chat # 根据可用模型选择如 deepseek-coder接下来创建config.py文件来加载配置并初始化模型。# config.py import os from dotenv import load_dotenv from langchain_openai import ChatOpenAI # 使用OpenAI兼容接口 # 加载.env文件中的环境变量 load_dotenv() def get_deepseek_llm(): 初始化并返回一个配置好的DeepSeek LLM实例。 由于DeepSeek API可能与OpenAI格式兼容我们使用ChatOpenAI并自定义base_url。 api_key os.getenv(DEEPSEEK_API_KEY) base_url os.getenv(DEEPSEEK_API_BASE) model_name os.getenv(DEEPSEEK_MODEL, deepseek-chat) if not api_key: raise ValueError(请在 .env 文件中设置 DEEPSEEK_API_KEY) llm ChatOpenAI( modelmodel_name, openai_api_keyapi_key, openai_api_basebase_url, temperature0.1, # 较低的温度使输出更确定适合任务执行 streamingFalse, # 非流式简化示例 timeout30, # 设置超时 ) return llm if __name__ __main__: # 简单测试连接 llm get_deepseek_llm() try: response llm.invoke(你好请用一句话介绍你自己。) print(连接测试成功) print(模型回复, response.content) except Exception as e: print(f连接失败{e})4.2 创建第一个工具并构建 ReAct Agent一个真正的 Agent 需要工具。我们创建一个简单的计算器和当前时间查询工具。# tools.py from datetime import datetime from langchain.tools import tool import math tool def calculator(expression: str) - str: 执行一个数学表达式计算。支持加减乘除(, -, *, /)和乘方(**)。 例如calculator(3 5 * 2) 或 calculator(sqrt(16))。 # 安全警告在生产环境中直接eval是危险的此处仅用于演示。 # 应使用更安全的表达式解析库如 ast.literal_eval 配合自定义解析。 try: # 为数学函数创建安全上下文 safe_dict {__builtins__: None} safe_dict.update(math.__dict__) # 允许使用math模块的函数如 sqrt, sin # 注意此方法仍有风险仅用于演示。真实项目请用更安全的方式。 result eval(expression, {__builtins__: None}, safe_dict) return f计算结果{expression} {result} except Exception as e: return f计算错误无法解析表达式 {expression}。错误信息{e} tool def get_current_time(timezone: str Asia/Shanghai) - str: 获取指定时区的当前日期和时间。 参数 timezone: 时区字符串例如 Asia/Shanghai, UTC, America/New_York。 from datetime import datetime import pytz # 需要安装 pip install pytz try: tz pytz.timezone(timezone) current_time datetime.now(tz) return f{timezone} 的当前时间是{current_time.strftime(%Y-%m-%d %H:%M:%S %Z%z)} except pytz.exceptions.UnknownTimeZoneError: return f错误未知时区 {timezone}。请使用有效的时区名称如 Asia/Shanghai。 # 注意使用pytz需要安装可以在requirements.txt中添加或运行 pip install pytz现在我们将模型、工具组合起来创建一个遵循 ReActReasoning Acting模式的 Agent。ReAct 是 Agent 的经典范式它让模型先“思考”Reasoning再“行动”Acting。# simple_agent.py from langchain.agents import create_react_agent, AgentExecutor from langchain import hub # 用于拉取预定义的提示词 from config import get_deepseek_llm from tools import calculator, get_current_time def create_simple_agent(): # 1. 初始化模型 llm get_deepseek_llm() # 2. 准备工具列表 tools [calculator, get_current_time] # 3. 从LangChain Hub拉取一个针对ReAct模式优化过的提示词模板 # 这个提示词会指导模型如何格式化它的“思考”和“行动” prompt hub.pull(hwchase17/react) # 4. 使用模型、工具和提示词创建ReAct Agent agent create_react_agent(llm, tools, prompt) # 5. 创建Agent执行器它负责运行Agent的循环思考-行动-观察-再思考... agent_executor AgentExecutor( agentagent, toolstools, verboseTrue, # 开启详细日志方便观察Agent的思考过程 handle_parsing_errorsTrue, # 优雅处理模型输出解析错误 max_iterations5, # 限制最大迭代次数防止无限循环 early_stopping_methodgenerate, # 当模型决定任务完成时停止 ) return agent_executor if __name__ __main__: agent create_simple_agent() # 测试几个问题 test_queries [ 现在上海是几点钟, 计算一下 15 的平方加上 20 除以 4 等于多少, 先告诉我现在的时间然后计算从2020年1月1日到今天过去了多少天提示可能需要更复杂的工具 ] for query in test_queries: print(f\n{*50}) print(f用户问题{query}) print(f{*50}) try: result agent.invoke({input: query}) print(f最终答案{result[output]}) except Exception as e: print(f执行出错{e})运行python simple_agent.py你会看到类似以下的详细输出它展示了 Agent 内部的思考链Chain of Thought 用户问题现在上海是几点钟 进入新的 Agent 执行链... 思考我需要找到上海当前的时间。我有一个工具可以获取指定时区的时间。 行动get_current_time 行动输入{timezone: Asia/Shanghai} 观察Asia/Shanghai 的当前时间是2023-10-27 14:30:15 CST0800 思考我已经得到了上海的时间可以回答用户了。 最终答案上海Asia/Shanghai的当前时间是 2023-10-27 14:30:15。这个简单的 Agent 已经具备了规划选择正确的工具和执行调用工具的能力。然而对于第三个更复杂的问题计算天数差我们现有的工具无法解决Agent 可能会在几次尝试后失败或给出错误答案。这引出了下一个话题如何设计更强大的工具和更稳健的流程这就是 Harness 工程要解决的问题。5. 引入 Harness 理念构建稳健的 Agent 工作流基础的AgentExecutor已经提供了很多功能但对于生产环境我们常常需要更精细的控制、状态管理和错误处理。LangGraph是一个基于图Graph来定义和运行 Agent 工作流的强大框架它完美体现了“Harness”的思想——为 Agent 套上可控的“缰绳”。5.1 使用 LangGraph 定义有状态的 Agent我们将重构之前的 Agent使用 LangGraph 来构建一个具有明确状态和节点的工作流。# graph_agent.py from typing import TypedDict, Annotated, List import operator from langgraph.graph import StateGraph, END from langgraph.prebuilt import ToolExecutor from langchain_core.messages import HumanMessage, AIMessage, ToolMessage from config import get_deepseek_llm from tools import calculator, get_current_time from langchain.tools.render import render_text_description # 将工具列表渲染为描述文本 # 1. 定义状态结构 class AgentState(TypedDict): 定义Agent工作流的状态。 messages: 存储所有的消息历史用户输入、AI回复、工具调用结果。 messages: Annotated[List, operator.add] # 这是一个特殊的注解表示该字段会追加内容 # 2. 初始化模型和工具 llm get_deepseek_llm() tools [calculator, get_current_time] tool_executor ToolExecutor(tools) # 工具执行器 # 3. 构建提示词比之前更详细地指导模型使用工具 def create_prompt(state: AgentState): # 获取对话历史 messages state[messages] # 将工具列表转换为模型能理解的描述字符串 tools_description render_text_description(tools) # 构建系统提示词 system_prompt f你是一个乐于助人的AI助手可以调用工具来解决问题。 你可以使用的工具如下 {tools_description} 调用工具时请严格按照以下格式 Action: 工具名称 Action Input: 工具的输入参数必须是有效的JSON字符串 当工具返回结果后我会以“Observation: ”开头提供结果给你。 你必须基于观察结果进行思考然后给出最终答案或继续调用下一个工具。 你的最终答案应以“Final Answer: ”开头。 # 返回完整的消息列表系统提示 历史消息 return [{role: system, content: system_prompt}] messages # 4. 定义工作流中的节点Nodes def call_model(state: AgentState): 调用大模型决定下一步是回复还是调用工具。 # 准备输入消息 prompt_messages create_prompt(state) # 调用模型并告诉它可以使用哪些工具bind_tools llm_with_tools llm.bind_tools(tools) response llm_with_tools.invoke(prompt_messages) # 将模型的响应添加到消息历史中 return {messages: [response]} def execute_tools(state: AgentState): 执行模型选择的工具。 last_message state[messages][-1] tool_calls last_message.tool_calls # 获取模型请求调用的工具列表 if not tool_calls: raise ValueError(没有需要执行的工具调用) results [] for tool_call in tool_calls: # 执行每一个工具调用 result tool_executor.invoke(tool_call) # 将工具执行结果封装为ToolMessage并关联到对应的tool_call_id results.append(ToolMessage(contentstr(result), tool_call_idtool_call[id])) # 将工具执行结果添加到消息历史 return {messages: results} # 5. 定义条件路由Edges def should_continue(state: AgentState) - str: 根据最后一条消息决定下一步是调用工具还是结束。 last_message state[messages][-1] # 如果最后一条消息是AIMessage且包含工具调用则去执行工具 if hasattr(last_message, tool_calls) and last_message.tool_calls: return call_tool # 否则工作流结束 return end # 6. 组装工作流图Graph workflow StateGraph(AgentState) # 添加节点 workflow.add_node(agent, call_model) # “思考”节点 workflow.add_node(action, execute_tools) # “行动”节点 # 设置入口点 workflow.set_entry_point(agent) # 添加条件边 workflow.add_conditional_edges( agent, should_continue, # 条件判断函数 { call_tool: action, # 如果需要调用工具则前往“action”节点 end: END # 否则结束 } ) # 从“行动”节点无条件返回“思考”节点形成循环 workflow.add_edge(action, agent) # 编译图 app workflow.compile() # 7. 运行工作流 if __name__ __main__: # 初始化状态包含用户的第一条消息 initial_state: AgentState { messages: [HumanMessage(content现在上海是几点钟然后计算当前小时数的平方。)] } print(开始运行 LangGraph Agent 工作流...) # 流式输出每一步的结果便于观察 for event in app.stream(initial_state, stream_modevalues): event_messages event.get(messages, []) if event_messages: last_message event_messages[-1] # 打印AI的思考/回复 if isinstance(last_message, AIMessage): print(f\n[AI思考] {last_message.content}) if last_message.tool_calls: print(f[AI决定调用工具] {[tc[name] for tc in last_message.tool_calls]}) # 打印工具执行结果 elif isinstance(last_message, ToolMessage): print(f[工具结果] {last_message.content}) # 打印用户输入只在开始时 elif isinstance(last_message, HumanMessage): print(f[用户] {last_message.content}) # 获取最终状态和答案 final_state app.invoke(initial_state) final_messages final_state[messages] # 提取最后一条AI消息作为最终答案 final_answer None for msg in reversed(final_messages): if isinstance(msg, AIMessage) and not msg.tool_calls: final_answer msg.content break print(f\n{*60}) print(f最终答案{final_answer})这个基于 LangGraph 的 Agent 工作流具有以下Harness 优势显式状态管理所有对话历史都清晰地在AgentState中维护。可控的工作流通过节点Nodes和边Edges明确定义了“思考-判断-行动-再思考”的循环。更好的可观测性我们可以轻松地在每个节点前后添加日志、监控或拦截逻辑。更强的错误处理能力可以在每个节点内部实现更精细的异常捕获和恢复机制。5.2 为工作流添加“安全护栏”GuardrailsHarness 的核心之一是控制。让我们为工具调用添加一个简单的安全检查层防止模型滥用危险工具如我们示例中使用了eval的calculator。# safety_harness.py import re from typing import Dict, Any class ToolSafetyHarness: 一个简单的工具安全套件示例。 staticmethod def sanitize_calculator_input(expression: str) - str: 对计算器输入进行简单的净化。 这是一个基础示例真实环境需要更严格的检查。 # 定义允许的字符集数字、基本运算符、括号、空格、小数点、math函数名 allowed_pattern r^[0-9\-*/().\s,]*$|^(sqrt|sin|cos|tan|log|exp)\([^)]*\)$ # 移除多余空格 expr_clean expression.strip() # 检查是否包含明显危险的字符串 dangerous_keywords [__, import, exec, eval, open, file, os., sys.] for keyword in dangerous_keywords: if keyword in expr_clean.lower(): raise ValueError(f输入包含潜在危险关键字: {keyword}) # 检查是否符合允许的模式简化检查 # 注意这是一个非常基础的检查不能完全保证安全。 if not re.match(r^[\d\-*/().\s,sqrt sincostanlogexp]$, expr_clean): # 如果基础检查不通过尝试匹配函数调用模式 if not re.match(r^(sqrt|sin|cos|tan|log|exp)\([\d\-*/().\s,]\)$, expr_clean): raise ValueError(f输入表达式格式不安全或不被支持: {expr_clean}) return expr_clean staticmethod def validate_timezone(timezone: str) - str: 验证时区字符串是否基本合规。 # 简单的时区格式检查例如Continent/City 格式 if not re.match(r^[A-Za-z]/[A-Za-z_]$, timezone): # 允许一些常见缩写 common_tz [UTC, GMT, EST, PST, CST] if timezone not in common_tz: raise ValueError(f时区格式可能无效: {timezone}。请使用类似 Asia/Shanghai 的格式。) return timezone def create_safe_calculator_tool(): 创建一个经过安全包装的计算器工具。 from langchain.tools import tool from tools import calculator as original_calculator harness ToolSafetyHarness() tool def safe_calculator(expression: str) - str: try: safe_expression harness.sanitize_calculator_input(expression) # 调用原始工具函数但传入净化后的输入 # 注意这里直接调用了原函数实际应重构原工具逻辑以避免eval。 # 更安全的方式是实现一个不使用eval的解析器。 return original_calculator.invoke(safe_expression) except ValueError as e: return f安全校验失败{e} except Exception as e: return f计算过程出错{e} return safe_calculator # 在 graph_agent.py 中我们可以用 safe_calculator 替换原来的 calculator # tools [create_safe_calculator_tool(), get_current_time]关键点这个安全层只是一个示例。在生产环境中对于像“计算器”这样执行代码的工具最佳实践是彻底避免eval使用安全的数学表达式解析库如asteval一个限制性的求值器。沙箱环境在隔离的容器或沙箱中执行不可信的代码。严格的输入白名单只允许预先定义好的、无害的操作和函数。6. 工程化扩展记忆、工具库与智能路由一个成熟的 Agent Harness 还需要解决更多工程问题。6.1 持久化记忆Memory让 Agent 记住跨会话的上下文。我们可以使用 LangChain 提供的记忆组件。# memory_agent.py from langgraph.graph import StateGraph, END from langgraph.checkpoint.memory import MemorySaver # 内存检查点保存器 from config import get_deepseek_llm from tools import calculator, get_current_time # ... 省略之前定义 AgentState, call_model, execute_tools, should_continue 的代码 ... def create_agent_with_memory(): llm get_deepseek_llm() tools [calculator, get_current_time] # 1. 创建检查点存储器这里使用内存生产环境可用数据库 memory MemorySaver() # 2. 构建工作流图同之前 workflow StateGraph(AgentState) workflow.add_node(agent, call_model) workflow.add_node(action, execute_tools) workflow.set_entry_point(agent) workflow.add_conditional_edges( agent, should_continue, {call_tool: action, end: END} ) workflow.add_edge(action, agent) # 3. 编译图时传入检查点存储器 app workflow.compile(checkpointermemory) return app, memory if __name__ __main__: app, memory create_agent_with_memory() # 模拟一个对话线程 thread_id user_123_session_1 config {configurable: {thread_id: thread_id}} # 第一轮对话 print( 第一轮对话 ) initial_state {messages: [HumanMessage(content我叫小明。)]} result1 app.invoke(initial_state, config) print(fAI: {result1[messages][-1].content}) # 第二轮对话记忆会保留 print(\n 第二轮对话 ) result2 app.invoke({messages: [HumanMessage(content我的名字是什么)]}, config) print(fAI: {result2[messages][-1].content}) # Agent 应该能回答“你叫小明”6.2 工具库与动态工具选择当工具很多时让模型每次都在所有工具中选择效率低下。我们可以根据用户问题先路由到不同的工具子集。# tool_router.py from langchain.tools import Tool from langchain.agents import create_tool_calling_agent from langchain.agents import AgentExecutor def create_tool_router_agent(): llm get_deepseek_llm() # 定义多个工具并为其添加描述和分类标签 math_tools [ Tool( nameadvanced_calculator, funccalculator, description用于执行数学表达式计算。输入应为字符串格式的数学表达式。, tags[math, calculation] ), ] time_tools [ Tool( nameworld_clock, funcget_current_time, description获取全球任何时区的当前时间。输入应为时区字符串如 Asia/Shanghai。, tags[time, utility] ), ] # 模拟一个搜索工具需要安装相关库如 duckduckgo-search from langchain_community.tools import DuckDuckGoSearchRun search_tool DuckDuckGoSearchRun() info_tools [ Tool( nameweb_search, funcsearch_tool.run, description在互联网上搜索最新信息。输入应为搜索查询关键词。, tags[search, information] ), ] # 根据问题类型选择工具集的简单路由逻辑实际可以使用一个分类模型 def route_tools(query: str) - list: query_lower query.lower() if any(word in query_lower for word in [计算, 等于, 加减, 乘除, 平方, 数学]): return math_tools elif any(word in query_lower for word in [时间, 几点, 时区, 钟表]): return time_tools else: # 默认返回搜索和信息类工具 return info_tools time_tools # 组合 # 这个Agent执行器可以根据每次的问题动态选择工具集 # 注意这是一个简化示例实际实现可能需要更复杂的路由机制。 class RoutingAgentExecutor: def __init__(self, llm): self.llm llm def invoke(self, input_data: dict): query input_data[input] selected_tools route_tools(query) # 为选中的工具集动态创建Agent prompt hub.pull(hwchase17/react) agent create_tool_calling_agent(self.llm, selected_tools, prompt) agent_executor AgentExecutor(agentagent, toolsselected_tools, verboseTrue) return agent_executor.invoke(input_data) return RoutingAgentExecutor(llm)7. 部署与监控建议将 DeepSeek Agent 投入生产环境Harness 工程还需要考虑以下方面7.1 部署模式API 服务化使用 FastAPI 或 Flask 将 Agent 包装成 RESTful API。# app.py (FastAPI示例) from fastapi import FastAPI, HTTPException from pydantic import BaseModel from graph_agent import app as agent_workflow # 导入之前编译好的LangGraph应用 app FastAPI(titleDeepSeek Agent API) class QueryRequest(BaseModel): question: str thread_id: str default_session app.post(/ask) async def ask_agent(request: QueryRequest): try: config {configurable: {thread_id: request.thread_id}} initial_state {messages: [HumanMessage(contentrequest.question)]} result agent_workflow.invoke(initial_state, config) # 提取最终答案 final_answer ... return {answer: final_answer, session_id: request.thread_id} except Exception as e: raise HTTPException(status_code500, detailstr(e))异步处理对于耗时任务使用 Celery 或 Dramatiq 进行异步任务队列处理。容器化使用 Docker 打包应用确保环境一致性。7.2 监控与可观测性日志记录结构化记录每个 Agent 调用的输入、输出、工具使用、耗时和 token 消耗。性能指标监控 API 响应时间、错误率、模型调用延迟。成本控制记录每次调用的 token 数设置预算和用量告警。对话质量评估可以抽样进行人工评估或利用另一个模型进行自动评分。7.3 配置管理将模型配置API Key, Base URL, 模型名称、工具开关、超时设置、迭代次数限制等抽取到外部配置文件如config.yaml或环境变量中。使用pydantic-settings等库进行强类型配置管理。8. 常见问题与排查思路在开发和运行 DeepSeek Agent 过程中你可能会遇到以下典型问题问题现象可能原因排查思路与解决方案API 调用失败返回认证错误1. API Key 错误或过期。2. API Base URL 不正确。3. 网络问题导致无法访问 DeepSeek 服务。1. 检查.env文件中的DEEPSEEK_API_KEY是否正确并在官方平台验证其有效性。2. 核对DEEPSEEK_API_BASE是否为官方提供的最新地址。3. 使用curl或requests库直接测试 API 连通性。Agent 陷入无限循环或多次调用工具1.max_iterations设置过高或未设置。2. 模型未能正确理解任务已完成。3. 工具返回的结果未能让模型做出结束判断。1. 在AgentExecutor或 LangGraph 工作流中明确设置max_iterations如 10。2. 优化系统提示词明确指示模型在得到答案后输出“Final Answer:”。3. 检查工具返回的结果是否清晰、格式正确。模型不调用工具直接回答问题1. 工具描述不够清晰模型不理解其用途。2. 提示词未有效激励模型使用工具。3. 模型温度 (temperature) 设置过高导致输出随机性大。1. 为每个工具编写详细、准确的description说明其用途、输入格式和输出示例。2. 使用专为工具调用设计的提示词模板如 LangChain Hub 中的hwchase17/react。3. 将temperature调低如 0.1使输出更确定。工具调用出错如计算器 eval 错误1. 工具函数内部代码异常。2. 模型生成的工具输入参数格式错误非 JSON。3. 工具输入不符合函数参数要求。1. 在工具函数内部添加完善的try...except异常捕获并返回友好的错误信息。2. 使用handle_parsing_errorsTrue参数让 AgentExecutor 能处理格式错误。3. 在工具描述中明确指定输入参数的类型和示例。LangGraph 工作流状态混乱1.AgentState定义不正确特别是Annotated字段。2. 节点函数没有返回正确的状态更新字典。3. 消息类型HumanMessage,AIMessage,ToolMessage使用错误。1. 仔细阅读 LangGraph 文档确保状态结构定义正确。operator.add用于列表追加。2. 确保每个节点函数都返回一个字典其键是状态字段名值是更新内容。3. 使用 LangChain 提供的标准消息类确保tool_calls等属性正确传递。部署后性能低下1. 网络延迟高。2. 未使用异步处理。3. 工具调用是同步阻塞的。1. 考虑将服务部署在离 DeepSeek API 服务器更近的区域。2. 对于 Web 服务使用异步框架如 FastAPI和异步的 LangChain 调用。3. 对于耗时的工具如网络请求将其改造成异步函数。9. 最佳实践与工程建议提示词工程是核心Agent 的表现极大程度上依赖于提示词。精心设计系统提示词明确角色、规则、工具使用格式和输出要求。可以准备多个提示词模板用于不同场景如数据分析、客服、编程辅助。工具设计要原子化且安全每个工具应只做一件事并做好做好。输入输出接口要清晰。对于执行代码、访问文件系统或网络资源的工具必须实施严格的安全检查、权限控制和沙箱机制。实施严格的输入验证与净化对所有来自用户输入和模型生成的内容特别是传递给工具的参数进行验证、转义和净化防止注入攻击。设置明确的边界与限制通过max_iterations、max_execution_time、token预算等机制防止 Agent 运行失控或产生过高成本。建立完整的测试套件为你的 Agent 编写单元测试测试单个工具、集成测试测试工具链和端到端测试测试完整对话流。使用包含边界案例和对抗性提示的测试集。版本化与管理配置对提示词、工具集、模型参数等所有配置进行版本控制如 Git。这便于回滚、对比实验和协作。规划可扩展的架构从一开始就考虑如何添加新工具。可以设计一个工具注册中心支持动态加载和卸载工具而无需重启服务。重视可观测性在关键节点模型调用、工具执行、最终输出记录详细的日志和指标。这不仅是调试的需要也是分析 Agent 行为、优化提示词和工具的基础。成本与性能优化对于复杂任务可以考虑让 Agent 先制定一个计划Plan然后并行执行其中不依赖的工具调用Action最后综合结果Synthesis。这能有效减少顺序调用带来的延迟。保持简洁逐步复杂化不要一开始就追求一个“全能”的 Agent。从一个解决特定问题的小型、稳健的 Agent 开始验证其价值再逐步扩展其能力和范围。构建一个真正强大、可靠的 DeepSeek Agent 并非一蹴而就它需要将强大的模型能力与严谨的软件工程实践即 Harness相结合。从明确的需求定义到安全的工具开发再到稳健的工作流编排和全面的生产部署每一步都至关重要。希望本文提供的概念解析、实战代码和工程建议能为你搭建自己的智能体应用提供一个坚实的起点。 基于DeepSeek构建AI Agent：从ReAct模式到LangGraph工程化实践
最近在探索大语言模型应用开发时发现一个现象很多开发者尝试将 DeepSeek 模型接入到各种 Agent 框架中但常常在环境配置、工具调用和流程编排上遇到瓶颈。与此同时一个名为 “Harness” 的概念在技术社区中频繁出现它似乎为 DeepSeek 的 Agent 化应用提供了一套更系统、更工程化的解决方案。本文将深入探讨如何基于 DeepSeek 模型构建一个功能完备的 AI Agent并重点解析 “Harness” 工程化实践的核心思想、技术实现与最佳路径。无论你是想快速搭建一个智能对话助手还是希望构建一个能处理复杂任务的企业级智能体本文提供的从零到一的完整指南和避坑方案都能为你提供直接可复用的参考。1. 背景与核心概念从 DeepSeek 到智能体工程在深入实战之前我们需要厘清几个关键概念这有助于理解整个技术栈的全貌。1.1 DeepSeek 模型强大的基座能力DeepSeek 是由深度求索公司开发的一系列大型语言模型。它因其在代码生成、逻辑推理和中文理解方面的出色表现而受到开发者社区的广泛关注。与一些通用聊天模型不同DeepSeek 的多个版本如 DeepSeek-Coder在编程任务上进行了深度优化使其成为构建开发辅助、自动化脚本生成等 Agent 的理想“大脑”。我们可以通过其提供的 API 来调用模型能力这是构建 Agent 的起点。1.2 AI Agent从“聊天”到“做事”AI Agent智能体不同于简单的聊天机器人。一个真正的 Agent 应具备以下核心能力感知与理解解析用户的自然语言指令理解其深层意图。规划与决策将复杂任务分解为可执行的子步骤序列。工具使用调用外部工具如搜索引擎、数据库、代码执行环境来获取信息或执行操作。记忆与学习在对话或任务执行过程中保持上下文并能从历史中学习。简单来说一个调用 DeepSeek API 的对话程序只是一个“问答机”而一个集成了工具调用、具备任务规划能力的 DeepSeek Agent 则是一个可以自主“做事”的智能助手。1.3 HarnessAgent 的工程化“缰绳”“Harness”在工程领域常指“线束”或“控制装置”。在 AI Agent 的语境下Harness 指的是一套用于控制、编排、测试和保障 Agent 稳定可靠运行的工程化框架和最佳实践集合。它解决了 Agent 开发中的常见痛点流程失控Agent 的思维链可能发散需要约束其行为边界。工具混乱多个工具如何被安全、高效地调用和管理。状态管理复杂的多轮对话和任务执行状态如何持久化和恢复。测试与评估如何系统化地测试 Agent 在各种场景下的表现。部署与监控如何将 Agent 部署到生产环境并监控其运行状态。你可以将 Harness 理解为 Agent 开发中的“Spring Framework”它提供了构建生产级智能体所需的基础设施和设计模式。网络上讨论的 “Harness Engineering” 或 “Agent Harness” 正是聚焦于这方面的工程实践。2. 环境准备与版本说明在开始构建我们的 DeepSeek Agent 之前需要准备好开发环境。以下配置是一个通用性较强的起点你可以根据自己的系统进行调整。核心环境要求操作系统Windows 10/11, macOS 10.15或主流 Linux 发行版如 Ubuntu 20.04。本文示例基于 Ubuntu 22.04。Python版本 3.8 - 3.11。推荐使用 3.9 或 3.10 以获得最佳的库兼容性。使用python --version检查。包管理工具pip通常随 Python 安装。DeepSeek API 密钥你需要访问 DeepSeek 平台并注册获取 API Key这是调用模型服务的凭证。代码编辑器VS Code、PyCharm 等任选。推荐 VS Code并安装 Python 扩展。项目初始化首先创建一个干净的项目目录并初始化虚拟环境这是管理 Python 项目依赖的最佳实践。# 创建项目目录 mkdir deepseek-agent-harness cd deepseek-agent-harness # 创建虚拟环境Python 3.9示例 python3.9 -m venv venv # 激活虚拟环境 # Linux/macOS source venv/bin/activate # Windows # venv\Scripts\activate # 升级pip pip install --upgrade pip3. 核心依赖与 Harness 框架选型构建一个具备 Harness 工程化特性的 Agent我们需要选择合适的库。我们将以LangChain和LangGraph作为核心框架因为它们提供了强大的 Agent 抽象和流程编排能力并且社区活跃生态丰富。安装核心依赖# 安装LangChain核心及社区工具包 pip install langchain langchain-community # 安装LangGraph用于构建有状态的、循环的Agent工作流 pip install langgraph # 安装DeepSeek的LangChain集成包如果官方提供或通用的OpenAI兼容层 # 由于DeepSeek API可能与OpenAI格式兼容我们通常使用openai库但需配置自定义base_url pip install openai # 安装用于网页搜索的工具依赖示例工具 pip install duckduckgo-search # 安装环境变量管理库 pip install python-dotenv版本说明与兼容性提示langchain和langgraph版本迭代较快本文示例基于langchain0.1.0和langgraph0.0.20的较新版本。如果遇到语法错误请查阅对应版本的官方文档。DeepSeek 的 API 端点可能更新请以官方最新文档为准。虚拟环境能有效隔离依赖避免与系统其他Python项目冲突务必在激活状态下进行后续操作。4. 构建基础 DeepSeek Agent让我们从构建一个最简单的、能调用 DeepSeek 模型并回答问题的 Agent 开始。4.1 配置模型访问首先在项目根目录创建.env文件来安全地存储你的 API 密钥切勿将密钥硬编码在代码中。# .env 文件 DEEPSEEK_API_KEYyour_deepseek_api_key_here DEEPSEEK_API_BASEhttps://api.deepseek.com/v1 # 请根据官方文档确认最新端点 DEEPSEEK_MODELdeepseek-chat # 根据可用模型选择如 deepseek-coder接下来创建config.py文件来加载配置并初始化模型。# config.py import os from dotenv import load_dotenv from langchain_openai import ChatOpenAI # 使用OpenAI兼容接口 # 加载.env文件中的环境变量 load_dotenv() def get_deepseek_llm(): 初始化并返回一个配置好的DeepSeek LLM实例。 由于DeepSeek API可能与OpenAI格式兼容我们使用ChatOpenAI并自定义base_url。 api_key os.getenv(DEEPSEEK_API_KEY) base_url os.getenv(DEEPSEEK_API_BASE) model_name os.getenv(DEEPSEEK_MODEL, deepseek-chat) if not api_key: raise ValueError(请在 .env 文件中设置 DEEPSEEK_API_KEY) llm ChatOpenAI( modelmodel_name, openai_api_keyapi_key, openai_api_basebase_url, temperature0.1, # 较低的温度使输出更确定适合任务执行 streamingFalse, # 非流式简化示例 timeout30, # 设置超时 ) return llm if __name__ __main__: # 简单测试连接 llm get_deepseek_llm() try: response llm.invoke(你好请用一句话介绍你自己。) print(连接测试成功) print(模型回复, response.content) except Exception as e: print(f连接失败{e})4.2 创建第一个工具并构建 ReAct Agent一个真正的 Agent 需要工具。我们创建一个简单的计算器和当前时间查询工具。# tools.py from datetime import datetime from langchain.tools import tool import math tool def calculator(expression: str) - str: 执行一个数学表达式计算。支持加减乘除(, -, *, /)和乘方(**)。 例如calculator(3 5 * 2) 或 calculator(sqrt(16))。 # 安全警告在生产环境中直接eval是危险的此处仅用于演示。 # 应使用更安全的表达式解析库如 ast.literal_eval 配合自定义解析。 try: # 为数学函数创建安全上下文 safe_dict {__builtins__: None} safe_dict.update(math.__dict__) # 允许使用math模块的函数如 sqrt, sin # 注意此方法仍有风险仅用于演示。真实项目请用更安全的方式。 result eval(expression, {__builtins__: None}, safe_dict) return f计算结果{expression} {result} except Exception as e: return f计算错误无法解析表达式 {expression}。错误信息{e} tool def get_current_time(timezone: str Asia/Shanghai) - str: 获取指定时区的当前日期和时间。 参数 timezone: 时区字符串例如 Asia/Shanghai, UTC, America/New_York。 from datetime import datetime import pytz # 需要安装 pip install pytz try: tz pytz.timezone(timezone) current_time datetime.now(tz) return f{timezone} 的当前时间是{current_time.strftime(%Y-%m-%d %H:%M:%S %Z%z)} except pytz.exceptions.UnknownTimeZoneError: return f错误未知时区 {timezone}。请使用有效的时区名称如 Asia/Shanghai。 # 注意使用pytz需要安装可以在requirements.txt中添加或运行 pip install pytz现在我们将模型、工具组合起来创建一个遵循 ReActReasoning Acting模式的 Agent。ReAct 是 Agent 的经典范式它让模型先“思考”Reasoning再“行动”Acting。# simple_agent.py from langchain.agents import create_react_agent, AgentExecutor from langchain import hub # 用于拉取预定义的提示词 from config import get_deepseek_llm from tools import calculator, get_current_time def create_simple_agent(): # 1. 初始化模型 llm get_deepseek_llm() # 2. 准备工具列表 tools [calculator, get_current_time] # 3. 从LangChain Hub拉取一个针对ReAct模式优化过的提示词模板 # 这个提示词会指导模型如何格式化它的“思考”和“行动” prompt hub.pull(hwchase17/react) # 4. 使用模型、工具和提示词创建ReAct Agent agent create_react_agent(llm, tools, prompt) # 5. 创建Agent执行器它负责运行Agent的循环思考-行动-观察-再思考... agent_executor AgentExecutor( agentagent, toolstools, verboseTrue, # 开启详细日志方便观察Agent的思考过程 handle_parsing_errorsTrue, # 优雅处理模型输出解析错误 max_iterations5, # 限制最大迭代次数防止无限循环 early_stopping_methodgenerate, # 当模型决定任务完成时停止 ) return agent_executor if __name__ __main__: agent create_simple_agent() # 测试几个问题 test_queries [ 现在上海是几点钟, 计算一下 15 的平方加上 20 除以 4 等于多少, 先告诉我现在的时间然后计算从2020年1月1日到今天过去了多少天提示可能需要更复杂的工具 ] for query in test_queries: print(f\n{*50}) print(f用户问题{query}) print(f{*50}) try: result agent.invoke({input: query}) print(f最终答案{result[output]}) except Exception as e: print(f执行出错{e})运行python simple_agent.py你会看到类似以下的详细输出它展示了 Agent 内部的思考链Chain of Thought 用户问题现在上海是几点钟 进入新的 Agent 执行链... 思考我需要找到上海当前的时间。我有一个工具可以获取指定时区的时间。 行动get_current_time 行动输入{timezone: Asia/Shanghai} 观察Asia/Shanghai 的当前时间是2023-10-27 14:30:15 CST0800 思考我已经得到了上海的时间可以回答用户了。 最终答案上海Asia/Shanghai的当前时间是 2023-10-27 14:30:15。这个简单的 Agent 已经具备了规划选择正确的工具和执行调用工具的能力。然而对于第三个更复杂的问题计算天数差我们现有的工具无法解决Agent 可能会在几次尝试后失败或给出错误答案。这引出了下一个话题如何设计更强大的工具和更稳健的流程这就是 Harness 工程要解决的问题。5. 引入 Harness 理念构建稳健的 Agent 工作流基础的AgentExecutor已经提供了很多功能但对于生产环境我们常常需要更精细的控制、状态管理和错误处理。LangGraph是一个基于图Graph来定义和运行 Agent 工作流的强大框架它完美体现了“Harness”的思想——为 Agent 套上可控的“缰绳”。5.1 使用 LangGraph 定义有状态的 Agent我们将重构之前的 Agent使用 LangGraph 来构建一个具有明确状态和节点的工作流。# graph_agent.py from typing import TypedDict, Annotated, List import operator from langgraph.graph import StateGraph, END from langgraph.prebuilt import ToolExecutor from langchain_core.messages import HumanMessage, AIMessage, ToolMessage from config import get_deepseek_llm from tools import calculator, get_current_time from langchain.tools.render import render_text_description # 将工具列表渲染为描述文本 # 1. 定义状态结构 class AgentState(TypedDict): 定义Agent工作流的状态。 messages: 存储所有的消息历史用户输入、AI回复、工具调用结果。 messages: Annotated[List, operator.add] # 这是一个特殊的注解表示该字段会追加内容 # 2. 初始化模型和工具 llm get_deepseek_llm() tools [calculator, get_current_time] tool_executor ToolExecutor(tools) # 工具执行器 # 3. 构建提示词比之前更详细地指导模型使用工具 def create_prompt(state: AgentState): # 获取对话历史 messages state[messages] # 将工具列表转换为模型能理解的描述字符串 tools_description render_text_description(tools) # 构建系统提示词 system_prompt f你是一个乐于助人的AI助手可以调用工具来解决问题。 你可以使用的工具如下 {tools_description} 调用工具时请严格按照以下格式 Action: 工具名称 Action Input: 工具的输入参数必须是有效的JSON字符串 当工具返回结果后我会以“Observation: ”开头提供结果给你。 你必须基于观察结果进行思考然后给出最终答案或继续调用下一个工具。 你的最终答案应以“Final Answer: ”开头。 # 返回完整的消息列表系统提示 历史消息 return [{role: system, content: system_prompt}] messages # 4. 定义工作流中的节点Nodes def call_model(state: AgentState): 调用大模型决定下一步是回复还是调用工具。 # 准备输入消息 prompt_messages create_prompt(state) # 调用模型并告诉它可以使用哪些工具bind_tools llm_with_tools llm.bind_tools(tools) response llm_with_tools.invoke(prompt_messages) # 将模型的响应添加到消息历史中 return {messages: [response]} def execute_tools(state: AgentState): 执行模型选择的工具。 last_message state[messages][-1] tool_calls last_message.tool_calls # 获取模型请求调用的工具列表 if not tool_calls: raise ValueError(没有需要执行的工具调用) results [] for tool_call in tool_calls: # 执行每一个工具调用 result tool_executor.invoke(tool_call) # 将工具执行结果封装为ToolMessage并关联到对应的tool_call_id results.append(ToolMessage(contentstr(result), tool_call_idtool_call[id])) # 将工具执行结果添加到消息历史 return {messages: results} # 5. 定义条件路由Edges def should_continue(state: AgentState) - str: 根据最后一条消息决定下一步是调用工具还是结束。 last_message state[messages][-1] # 如果最后一条消息是AIMessage且包含工具调用则去执行工具 if hasattr(last_message, tool_calls) and last_message.tool_calls: return call_tool # 否则工作流结束 return end # 6. 组装工作流图Graph workflow StateGraph(AgentState) # 添加节点 workflow.add_node(agent, call_model) # “思考”节点 workflow.add_node(action, execute_tools) # “行动”节点 # 设置入口点 workflow.set_entry_point(agent) # 添加条件边 workflow.add_conditional_edges( agent, should_continue, # 条件判断函数 { call_tool: action, # 如果需要调用工具则前往“action”节点 end: END # 否则结束 } ) # 从“行动”节点无条件返回“思考”节点形成循环 workflow.add_edge(action, agent) # 编译图 app workflow.compile() # 7. 运行工作流 if __name__ __main__: # 初始化状态包含用户的第一条消息 initial_state: AgentState { messages: [HumanMessage(content现在上海是几点钟然后计算当前小时数的平方。)] } print(开始运行 LangGraph Agent 工作流...) # 流式输出每一步的结果便于观察 for event in app.stream(initial_state, stream_modevalues): event_messages event.get(messages, []) if event_messages: last_message event_messages[-1] # 打印AI的思考/回复 if isinstance(last_message, AIMessage): print(f\n[AI思考] {last_message.content}) if last_message.tool_calls: print(f[AI决定调用工具] {[tc[name] for tc in last_message.tool_calls]}) # 打印工具执行结果 elif isinstance(last_message, ToolMessage): print(f[工具结果] {last_message.content}) # 打印用户输入只在开始时 elif isinstance(last_message, HumanMessage): print(f[用户] {last_message.content}) # 获取最终状态和答案 final_state app.invoke(initial_state) final_messages final_state[messages] # 提取最后一条AI消息作为最终答案 final_answer None for msg in reversed(final_messages): if isinstance(msg, AIMessage) and not msg.tool_calls: final_answer msg.content break print(f\n{*60}) print(f最终答案{final_answer})这个基于 LangGraph 的 Agent 工作流具有以下Harness 优势显式状态管理所有对话历史都清晰地在AgentState中维护。可控的工作流通过节点Nodes和边Edges明确定义了“思考-判断-行动-再思考”的循环。更好的可观测性我们可以轻松地在每个节点前后添加日志、监控或拦截逻辑。更强的错误处理能力可以在每个节点内部实现更精细的异常捕获和恢复机制。5.2 为工作流添加“安全护栏”GuardrailsHarness 的核心之一是控制。让我们为工具调用添加一个简单的安全检查层防止模型滥用危险工具如我们示例中使用了eval的calculator。# safety_harness.py import re from typing import Dict, Any class ToolSafetyHarness: 一个简单的工具安全套件示例。 staticmethod def sanitize_calculator_input(expression: str) - str: 对计算器输入进行简单的净化。 这是一个基础示例真实环境需要更严格的检查。 # 定义允许的字符集数字、基本运算符、括号、空格、小数点、math函数名 allowed_pattern r^[0-9\-*/().\s,]*$|^(sqrt|sin|cos|tan|log|exp)\([^)]*\)$ # 移除多余空格 expr_clean expression.strip() # 检查是否包含明显危险的字符串 dangerous_keywords [__, import, exec, eval, open, file, os., sys.] for keyword in dangerous_keywords: if keyword in expr_clean.lower(): raise ValueError(f输入包含潜在危险关键字: {keyword}) # 检查是否符合允许的模式简化检查 # 注意这是一个非常基础的检查不能完全保证安全。 if not re.match(r^[\d\-*/().\s,sqrt sincostanlogexp]$, expr_clean): # 如果基础检查不通过尝试匹配函数调用模式 if not re.match(r^(sqrt|sin|cos|tan|log|exp)\([\d\-*/().\s,]\)$, expr_clean): raise ValueError(f输入表达式格式不安全或不被支持: {expr_clean}) return expr_clean staticmethod def validate_timezone(timezone: str) - str: 验证时区字符串是否基本合规。 # 简单的时区格式检查例如Continent/City 格式 if not re.match(r^[A-Za-z]/[A-Za-z_]$, timezone): # 允许一些常见缩写 common_tz [UTC, GMT, EST, PST, CST] if timezone not in common_tz: raise ValueError(f时区格式可能无效: {timezone}。请使用类似 Asia/Shanghai 的格式。) return timezone def create_safe_calculator_tool(): 创建一个经过安全包装的计算器工具。 from langchain.tools import tool from tools import calculator as original_calculator harness ToolSafetyHarness() tool def safe_calculator(expression: str) - str: try: safe_expression harness.sanitize_calculator_input(expression) # 调用原始工具函数但传入净化后的输入 # 注意这里直接调用了原函数实际应重构原工具逻辑以避免eval。 # 更安全的方式是实现一个不使用eval的解析器。 return original_calculator.invoke(safe_expression) except ValueError as e: return f安全校验失败{e} except Exception as e: return f计算过程出错{e} return safe_calculator # 在 graph_agent.py 中我们可以用 safe_calculator 替换原来的 calculator # tools [create_safe_calculator_tool(), get_current_time]关键点这个安全层只是一个示例。在生产环境中对于像“计算器”这样执行代码的工具最佳实践是彻底避免eval使用安全的数学表达式解析库如asteval一个限制性的求值器。沙箱环境在隔离的容器或沙箱中执行不可信的代码。严格的输入白名单只允许预先定义好的、无害的操作和函数。6. 工程化扩展记忆、工具库与智能路由一个成熟的 Agent Harness 还需要解决更多工程问题。6.1 持久化记忆Memory让 Agent 记住跨会话的上下文。我们可以使用 LangChain 提供的记忆组件。# memory_agent.py from langgraph.graph import StateGraph, END from langgraph.checkpoint.memory import MemorySaver # 内存检查点保存器 from config import get_deepseek_llm from tools import calculator, get_current_time # ... 省略之前定义 AgentState, call_model, execute_tools, should_continue 的代码 ... def create_agent_with_memory(): llm get_deepseek_llm() tools [calculator, get_current_time] # 1. 创建检查点存储器这里使用内存生产环境可用数据库 memory MemorySaver() # 2. 构建工作流图同之前 workflow StateGraph(AgentState) workflow.add_node(agent, call_model) workflow.add_node(action, execute_tools) workflow.set_entry_point(agent) workflow.add_conditional_edges( agent, should_continue, {call_tool: action, end: END} ) workflow.add_edge(action, agent) # 3. 编译图时传入检查点存储器 app workflow.compile(checkpointermemory) return app, memory if __name__ __main__: app, memory create_agent_with_memory() # 模拟一个对话线程 thread_id user_123_session_1 config {configurable: {thread_id: thread_id}} # 第一轮对话 print( 第一轮对话 ) initial_state {messages: [HumanMessage(content我叫小明。)]} result1 app.invoke(initial_state, config) print(fAI: {result1[messages][-1].content}) # 第二轮对话记忆会保留 print(\n 第二轮对话 ) result2 app.invoke({messages: [HumanMessage(content我的名字是什么)]}, config) print(fAI: {result2[messages][-1].content}) # Agent 应该能回答“你叫小明”6.2 工具库与动态工具选择当工具很多时让模型每次都在所有工具中选择效率低下。我们可以根据用户问题先路由到不同的工具子集。# tool_router.py from langchain.tools import Tool from langchain.agents import create_tool_calling_agent from langchain.agents import AgentExecutor def create_tool_router_agent(): llm get_deepseek_llm() # 定义多个工具并为其添加描述和分类标签 math_tools [ Tool( nameadvanced_calculator, funccalculator, description用于执行数学表达式计算。输入应为字符串格式的数学表达式。, tags[math, calculation] ), ] time_tools [ Tool( nameworld_clock, funcget_current_time, description获取全球任何时区的当前时间。输入应为时区字符串如 Asia/Shanghai。, tags[time, utility] ), ] # 模拟一个搜索工具需要安装相关库如 duckduckgo-search from langchain_community.tools import DuckDuckGoSearchRun search_tool DuckDuckGoSearchRun() info_tools [ Tool( nameweb_search, funcsearch_tool.run, description在互联网上搜索最新信息。输入应为搜索查询关键词。, tags[search, information] ), ] # 根据问题类型选择工具集的简单路由逻辑实际可以使用一个分类模型 def route_tools(query: str) - list: query_lower query.lower() if any(word in query_lower for word in [计算, 等于, 加减, 乘除, 平方, 数学]): return math_tools elif any(word in query_lower for word in [时间, 几点, 时区, 钟表]): return time_tools else: # 默认返回搜索和信息类工具 return info_tools time_tools # 组合 # 这个Agent执行器可以根据每次的问题动态选择工具集 # 注意这是一个简化示例实际实现可能需要更复杂的路由机制。 class RoutingAgentExecutor: def __init__(self, llm): self.llm llm def invoke(self, input_data: dict): query input_data[input] selected_tools route_tools(query) # 为选中的工具集动态创建Agent prompt hub.pull(hwchase17/react) agent create_tool_calling_agent(self.llm, selected_tools, prompt) agent_executor AgentExecutor(agentagent, toolsselected_tools, verboseTrue) return agent_executor.invoke(input_data) return RoutingAgentExecutor(llm)7. 部署与监控建议将 DeepSeek Agent 投入生产环境Harness 工程还需要考虑以下方面7.1 部署模式API 服务化使用 FastAPI 或 Flask 将 Agent 包装成 RESTful API。# app.py (FastAPI示例) from fastapi import FastAPI, HTTPException from pydantic import BaseModel from graph_agent import app as agent_workflow # 导入之前编译好的LangGraph应用 app FastAPI(titleDeepSeek Agent API) class QueryRequest(BaseModel): question: str thread_id: str default_session app.post(/ask) async def ask_agent(request: QueryRequest): try: config {configurable: {thread_id: request.thread_id}} initial_state {messages: [HumanMessage(contentrequest.question)]} result agent_workflow.invoke(initial_state, config) # 提取最终答案 final_answer ... return {answer: final_answer, session_id: request.thread_id} except Exception as e: raise HTTPException(status_code500, detailstr(e))异步处理对于耗时任务使用 Celery 或 Dramatiq 进行异步任务队列处理。容器化使用 Docker 打包应用确保环境一致性。7.2 监控与可观测性日志记录结构化记录每个 Agent 调用的输入、输出、工具使用、耗时和 token 消耗。性能指标监控 API 响应时间、错误率、模型调用延迟。成本控制记录每次调用的 token 数设置预算和用量告警。对话质量评估可以抽样进行人工评估或利用另一个模型进行自动评分。7.3 配置管理将模型配置API Key, Base URL, 模型名称、工具开关、超时设置、迭代次数限制等抽取到外部配置文件如config.yaml或环境变量中。使用pydantic-settings等库进行强类型配置管理。8. 常见问题与排查思路在开发和运行 DeepSeek Agent 过程中你可能会遇到以下典型问题问题现象可能原因排查思路与解决方案API 调用失败返回认证错误1. API Key 错误或过期。2. API Base URL 不正确。3. 网络问题导致无法访问 DeepSeek 服务。1. 检查.env文件中的DEEPSEEK_API_KEY是否正确并在官方平台验证其有效性。2. 核对DEEPSEEK_API_BASE是否为官方提供的最新地址。3. 使用curl或requests库直接测试 API 连通性。Agent 陷入无限循环或多次调用工具1.max_iterations设置过高或未设置。2. 模型未能正确理解任务已完成。3. 工具返回的结果未能让模型做出结束判断。1. 在AgentExecutor或 LangGraph 工作流中明确设置max_iterations如 10。2. 优化系统提示词明确指示模型在得到答案后输出“Final Answer:”。3. 检查工具返回的结果是否清晰、格式正确。模型不调用工具直接回答问题1. 工具描述不够清晰模型不理解其用途。2. 提示词未有效激励模型使用工具。3. 模型温度 (temperature) 设置过高导致输出随机性大。1. 为每个工具编写详细、准确的description说明其用途、输入格式和输出示例。2. 使用专为工具调用设计的提示词模板如 LangChain Hub 中的hwchase17/react。3. 将temperature调低如 0.1使输出更确定。工具调用出错如计算器 eval 错误1. 工具函数内部代码异常。2. 模型生成的工具输入参数格式错误非 JSON。3. 工具输入不符合函数参数要求。1. 在工具函数内部添加完善的try...except异常捕获并返回友好的错误信息。2. 使用handle_parsing_errorsTrue参数让 AgentExecutor 能处理格式错误。3. 在工具描述中明确指定输入参数的类型和示例。LangGraph 工作流状态混乱1.AgentState定义不正确特别是Annotated字段。2. 节点函数没有返回正确的状态更新字典。3. 消息类型HumanMessage,AIMessage,ToolMessage使用错误。1. 仔细阅读 LangGraph 文档确保状态结构定义正确。operator.add用于列表追加。2. 确保每个节点函数都返回一个字典其键是状态字段名值是更新内容。3. 使用 LangChain 提供的标准消息类确保tool_calls等属性正确传递。部署后性能低下1. 网络延迟高。2. 未使用异步处理。3. 工具调用是同步阻塞的。1. 考虑将服务部署在离 DeepSeek API 服务器更近的区域。2. 对于 Web 服务使用异步框架如 FastAPI和异步的 LangChain 调用。3. 对于耗时的工具如网络请求将其改造成异步函数。9. 最佳实践与工程建议提示词工程是核心Agent 的表现极大程度上依赖于提示词。精心设计系统提示词明确角色、规则、工具使用格式和输出要求。可以准备多个提示词模板用于不同场景如数据分析、客服、编程辅助。工具设计要原子化且安全每个工具应只做一件事并做好做好。输入输出接口要清晰。对于执行代码、访问文件系统或网络资源的工具必须实施严格的安全检查、权限控制和沙箱机制。实施严格的输入验证与净化对所有来自用户输入和模型生成的内容特别是传递给工具的参数进行验证、转义和净化防止注入攻击。设置明确的边界与限制通过max_iterations、max_execution_time、token预算等机制防止 Agent 运行失控或产生过高成本。建立完整的测试套件为你的 Agent 编写单元测试测试单个工具、集成测试测试工具链和端到端测试测试完整对话流。使用包含边界案例和对抗性提示的测试集。版本化与管理配置对提示词、工具集、模型参数等所有配置进行版本控制如 Git。这便于回滚、对比实验和协作。规划可扩展的架构从一开始就考虑如何添加新工具。可以设计一个工具注册中心支持动态加载和卸载工具而无需重启服务。重视可观测性在关键节点模型调用、工具执行、最终输出记录详细的日志和指标。这不仅是调试的需要也是分析 Agent 行为、优化提示词和工具的基础。成本与性能优化对于复杂任务可以考虑让 Agent 先制定一个计划Plan然后并行执行其中不依赖的工具调用Action最后综合结果Synthesis。这能有效减少顺序调用带来的延迟。保持简洁逐步复杂化不要一开始就追求一个“全能”的 Agent。从一个解决特定问题的小型、稳健的 Agent 开始验证其价值再逐步扩展其能力和范围。构建一个真正强大、可靠的 DeepSeek Agent 并非一蹴而就它需要将强大的模型能力与严谨的软件工程实践即 Harness相结合。从明确的需求定义到安全的工具开发再到稳健的工作流编排和全面的生产部署每一步都至关重要。希望本文提供的概念解析、实战代码和工程建议能为你搭建自己的智能体应用提供一个坚实的起点。