DeepSeek Agent 开发教程

对比维度	传统 Chatbot	AI Agent
交互方式	单轮问答，一问一答	多步推理，自主执行任务链
能力边界	仅基于训练数据生成文本	可调用外部工具、API、数据库
记忆能力	上下文窗口内临时记忆	短期记忆 + 长期记忆（向量存储）
任务复杂度	简单对话，信息查询	多步骤任务，工作流自动化
典型场景	客服问答、闲聊、内容生成	数据分析、自动报告、代码执行、流程编排

组件	英文	作用	技术实现
大语言模型	LLM	Agent 的大脑，负责推理和决策	DeepSeek-Chat / DeepSeek-Reasoner
工具	Tools	Agent 的手脚，执行具体操作	Function Calling / API / 代码执行器
记忆	Memory	存储和检索历史信息	对话历史 / 向量数据库 / 知识图谱
规划	Planning	分解任务，制定执行策略	ReAct / Plan-and-Execute / Tree-of-Thought

1. 感知（Perception）：接收用户输入，理解任务意图和目标
2. 思考（Thinking）：分析任务，制定执行计划，决定下一步行动
3. 行动（Action）：调用工具执行具体操作（搜索、计算、API 调用等）
4. 观察（Observation）：获取工具执行结果，评估是否达成目标
5. 循环（Loop）：根据观察结果决定继续执行还是输出最终答案

步骤	说明	示例
Thought	分析当前状态，决定下一步做什么	"我需要查询今天的天气，然后决定是否建议带伞"
Action	执行具体的工具调用	调用 get_weather("北京") 函数
Observation	获取工具返回结果，评估是否完成任务	"北京今天晴，气温 25 度，无需带伞"

"""DeepSeek ReAct Agent — 完整实现""" import json import re from openai import OpenAI client = OpenAI( api_key="sk-your-api-key", base_url="https://api.deepseek.com/v1", ) # ========== 工具定义 ========== def search_web(query: str) -> str: """模拟搜索引擎，返回查询结果""" # 实际项目中替换为真实搜索 API mock_results = { "北京天气": "北京今天晴，气温 22-28 度，空气质量良", "DeepSeek": "DeepSeek 是深度求索公司开发的大语言模型，支持 1M 上下文", } return mock_results.get(query, f"未找到关于 '{query}' 的相关信息") def calculate(expression: str) -> str: """安全的数学计算器""" try: # 只允许数字和基本运算符 if not re.match(r'^[\d\+\-\*\/\(\)\.\s]+$', expression): return "错误: 表达式包含非法字符" result = eval(expression) return str(result) except Exception as e: return f"计算错误: {e}" # 工具注册表 TOOLS = { "search": search_web, "calculate": calculate, } # ========== ReAct Prompt ========== REACT_SYSTEM_PROMPT = """你是一个能够使用工具的智能 Agent。请严格按照 ReAct 格式回答。 可用工具： - search(query): 搜索互联网信息 - calculate(expression): 执行数学计算 回答格式（必须严格遵守）： Thought: 你对当前情况的思考和分析 Action: 要执行的操作，格式为 tool_name(arg) Observation: 工具执行的结果 ... （重复 Thought/Action/Observation 直到得出答案） Thought: 我已经有足够的信息来回答 Final Answer: 最终答案 注意：每次只能执行一个 Action，等待 Observation 后再继续。""" # ========== ReAct 循环 ========== def react_agent(user_query: str, max_steps: int = 5): """ReAct Agent 主循环""" messages = [ {"role": "system", "content": REACT_SYSTEM_PROMPT}, {"role": "user", "content": user_query}, ] for step in range(max_steps): print(f"\n--- Step {step + 1} ---") # 调用 DeepSeek 模型 response = client.chat.completions.create( model="deepseek-chat", messages=messages, temperature=0.1, ) reply = response.choices[0].message.content print(reply) messages.append({"role": "assistant", "content": reply}) # 检查是否是最终答案 if "Final Answer:" in reply: final = reply.split("Final Answer:")[-1].strip() return final # 解析 Action action_match = re.search(r'Action:\s*(\w+)\(([^)]*)\)', reply) if action_match: tool_name = action_match.group(1) tool_arg = action_match.group(2).strip().strip('"').strip("'") if tool_name in TOOLS: observation = TOOLS[tool_name](tool_arg) print(f"Observation: {observation}") messages.append({ "role": "user", "content": f"Observation: {observation}", }) else: messages.append({ "role": "user", "content": f"错误: 未知工具 '{tool_name}'", }) return "达到最大步数限制，未能完成任务。" # ========== 运行测试 ========== if __name__ == "__main__": query = "北京今天天气怎么样？如果气温超过 25 度，帮我算一下 30 个人每人需要 2 瓶水，一共需要多少瓶？" result = react_agent(query) print(f"\n{'='*60}\n最终答案: {result}\n{'='*60}")

ReAct 模式的关键要点

• 每次只执行一个 Action，等待 Observation 后再继续
• Thought 和 Action 必须严格遵守格式，便于正则解析
• 设置 max_steps 防止无限循环，通常 5-10 步足够
• 工具函数需要做好错误处理，返回有意义的结果
• DeepSeek 的推理能力非常适合 ReAct 模式

from openai import OpenAI import json client = OpenAI( api_key="sk-your-api-key", base_url="https://api.deepseek.com/v1", ) # 定义工具 Schema（OpenAI Function Calling 格式） tools = [ { "type": "function", "function": { "name": "get_weather", "description": "获取指定城市的实时天气信息", "parameters": { "type": "object", "properties": { "city": { "type": "string", "description": "城市名称，如 北京、上海、东京", }, "unit": { "type": "string", "enum": ["celsius", "fahrenheit"], "description": "温度单位，默认为 celsius", }, }, "required": ["city"], }, }, }, { "type": "function", "function": { "name": "search_database", "description": "在内部知识库中搜索信息", "parameters": { "type": "object", "properties": { "query": { "type": "string", "description": "搜索关键词或问题", }, "top_k": { "type": "integer", "description": "返回结果数量，默认 3", "default": 3, }, }, "required": ["query"], }, }, }, { "type": "function", "function": { "name": "send_email", "description": "发送电子邮件", "parameters": { "type": "object", "properties": { "to": {"type": "string", "description": "收件人邮箱地址"}, "subject": {"type": "string", "description": "邮件主题"}, "body": {"type": "string", "description": "邮件正文内容"}, }, "required": ["to", "subject", "body"], }, }, }, ]

# 工具实现 def get_weather(city: str, unit: str = "celsius") -> dict: """模拟天气查询（实际项目替换为真实 API）""" return { "city": city, "temperature": 26, "condition": "晴", "humidity": "45%", "unit": unit, } def search_database(query: str, top_k: int = 3) -> dict: return {"query": query, "results": [f"结果{i+1}: 关于 {query} 的信息" for i in range(top_k)]} def send_email(to: str, subject: str, body: str) -> dict: return {"status": "sent", "to": to, "subject": subject} # 工具映射表 AVAILABLE_FUNCTIONS = { "get_weather": get_weather, "search_database": search_database, "send_email": send_email, } # ========== Function Calling Agent ========== def function_calling_agent(user_query: str, max_turns: int = 5): messages = [ {"role": "system", "content": "你是一个智能助手，可以使用工具来完成任务。请用中文回答。"}, {"role": "user", "content": user_query}, ] for turn in range(max_turns): response = client.chat.completions.create( model="deepseek-chat", messages=messages, tools=tools, tool_choice="auto", # 模型自动决定是否调用工具 ) assistant_message = response.choices[0].message # 如果模型调用工具 if assistant_message.tool_calls: messages.append(assistant_message) for tool_call in assistant_message.tool_calls: func_name = tool_call.function.name func_args = json.loads(tool_call.function.arguments) print(f"[调用工具] {func_name}({func_args})") if func_name in AVAILABLE_FUNCTIONS: result = AVAILABLE_FUNCTIONS[func_name](**func_args) else: result = {"error": f"未知工具: {func_name}"} print(f"[工具结果] {result}") # 将工具结果加入对话 messages.append({ "role": "tool", "tool_call_id": tool_call.id, "content": json.dumps(result, ensure_ascii=False), }) else: # 模型直接返回答案 return assistant_message.content return "达到最大轮次限制。" # 测试 result = function_calling_agent("查询北京天气，然后给 admin@company.com 发一封天气日报") print(f"\n最终结果:\n{result}")

Function Calling 优势

• 模型自动判断何时需要调用工具，无需手动解析格式
• 参数以结构化 JSON 传递，避免正则解析错误
• 支持并行调用多个工具（tool_choice 设为 "auto"）
• DeepSeek API 完全兼容 OpenAI Function Calling 格式
• 可以混合使用：模型可在一次回复中同时调用工具和生成文本

pip install langchain langchain-openai langchain-community

from langchain_openai import ChatOpenAI from langchain.agents import create_react_agent, AgentExecutor from langchain.tools import Tool from langchain.prompts import PromptTemplate from langchain_community.tools import WikipediaQueryRun from langchain_community.utilities import WikipediaAPIWrapper import requests # 初始化 DeepSeek 模型 llm = ChatOpenAI( model="deepseek-chat", base_url="https://api.deepseek.com/v1", api_key="sk-your-api-key", temperature=0.1, ) # ========== 自定义工具开发 ========== # 工具 1: 汇率查询 def get_exchange_rate(currency_pair: str) -> str: """查询实时汇率。输入格式: USD/CNY 或 EUR/USD""" try: base, target = currency_pair.split("/") url = f"https://api.exchangerate-api.com/v4/latest/{base}" data = requests.get(url, timeout=5).json() rate = data["rates"][target] return f"1 {base} = {rate} {target}" except Exception as e: return f"汇率查询失败: {e}" # 工具 2: 文件读取 def read_file(filepath: str) -> str: """读取本地文件内容""" try: with open(filepath, "r", encoding="utf-8") as f: content = f.read() if len(content) > 2000: return content[:2000] + "\n... (内容过长，已截断)" return content except Exception as e: return f"读取文件失败: {e}" # 工具 3: Python 代码执行 def execute_python(code: str) -> str: """安全执行 Python 代码并返回结果""" # 注意: 生产环境应使用沙箱环境（如 Docker） import io, sys old_stdout = sys.stdout sys.stdout = buffer = io.StringIO() try: exec(code, {"__builtins__": { "print": print, "range": range, "len": len, "int": int, "str": str, "list": list, "dict": dict, "sum": sum, "sorted": sorted, "enumerate": enumerate, }}) result = buffer.getvalue() return result.strip() if result.strip() else "代码执行成功（无输出）" except Exception as e: return f"执行错误: {e}" finally: sys.stdout = old_stdout # ========== 注册工具 ========== tools = [ Tool( name="汇率查询", func=get_exchange_rate, description="查询实时汇率。输入格式: USD/CNY 或 EUR/USD", ), Tool( name="文件读取", func=read_file, description="读取本地文件内容。输入为文件路径", ), Tool( name="Python执行", func=execute_python, description="执行 Python 代码并返回结果。输入为 Python 代码字符串", ), Tool( name="维基百科", func=WikipediaQueryRun(api_wrapper=WikipediaAPIWrapper(lang="zh")).run, description="搜索中文维基百科。输入为搜索关键词", ), ] # ReAct Prompt 模板 prompt = PromptTemplate.from_template("""你是一个智能助手，可以使用以下工具完成任务： {tools} 使用以下格式回答： Question: 用户的问题 Thought: 思考下一步该做什么 Action: 工具名称 Action Input: 工具输入参数 Observation: 工具返回结果 ... (可重复 Thought/Action/Action Input/Observation) Thought: 我已经有足够的信息 Final Answer: 最终答案 开始！ Question: {input} Thought: {agent_scratchpad}""") # 创建 Agent 和执行器 agent = create_react_agent(llm=llm, tools=tools, prompt=prompt) agent_executor = AgentExecutor( agent=agent, tools=tools, verbose=True, # 打印详细执行过程 handle_parsing_errors=True, # 自动处理格式错误 max_iterations=8, # 最大迭代次数 ) # 运行 Agent result = agent_executor.invoke({ "input": "查询 USD/CNY 汇率，然后计算 1000 美元能换多少人民币", }) print(f"\n最终答案: {result['output']}")

参数	说明	推荐值
verbose	是否打印详细执行过程	开发时 True，生产时 False
handle_parsing_errors	自动处理模型输出格式错误	True
max_iterations	最大迭代次数，防止无限循环	5-10
early_stopping_method	达到最大迭代后的处理方式	"generate"（生成最终答案）

• Agent（智能体）：定义角色、目标、背景故事和可用工具
• Task（任务）：定义要完成的具体工作，包括描述、期望输出和分配的 Agent
• Crew（团队）：组织多个 Agent 按顺序或层级协作完成一组任务
• Process（流程）：控制任务执行顺序，支持 sequential（顺序）和 hierarchical（层级）

# pip install crewai crewai-tools from crewai import Agent, Task, Crew, Process from crewai import LLM # 配置 DeepSeek 作为底层模型 deepseek_llm = LLM( model="deepseek/deepseek-chat", base_url="https://api.deepseek.com/v1", api_key="sk-your-api-key", ) # ========== 定义 Agent 角色 ========== # 研究员: 负责信息收集和分析 researcher = Agent( role="高级研究员", goal="深入调研指定主题，收集全面准确的信息和数据", backstory="""你是一位经验丰富的研究分析师，曾在顶级咨询公司工作 10 年。 你擅长快速搜集信息、识别关键趋势、用数据支撑观点。 你的报告总是结构清晰、论据充分。""", llm=deepseek_llm, verbose=True, allow_delegation=False, ) # 写手: 负责内容创作 writer = Agent( role="资深内容写手", goal="基于研究资料撰写引人入胜、专业准确的内容", backstory="""你是一位资深科技写手，曾为多家知名科技媒体供稿。 你擅长将复杂的技术概念转化为通俗易懂的文章。 你的文章逻辑清晰、数据准确、可读性强。""", llm=deepseek_llm, verbose=True, allow_delegation=False, ) # 审核员: 负责质量把关 reviewer = Agent( role="内容审核员", goal="审核内容的准确性、可读性和专业性，确保输出质量", backstory="""你是一位资深编辑，曾在顶级出版社工作 15 年。 你对文字有极高的要求，能发现最细微的错误。 你审核的标准包括：事实准确性、逻辑连贯性、语言表达和格式规范。""", llm=deepseek_llm, verbose=True, allow_delegation=False, ) # ========== 定义任务 ========== research_task = Task( description="""研究 DeepSeek Agent 开发的最新进展，包括： 1. DeepSeek API 的 Function Calling 能力 2. LangChain 和 CrewAI 与 DeepSeek 的集成方式 3. Agent 在生产环境中的最佳实践 4. 与其他 LLM（GPT-4、Claude）的 Agent 能力对比 请提供详细的数据和具体的技术细节。""", expected_output="一份结构化的研究报告，包含关键发现、技术对比和最佳实践建议，不少于 500 字", agent=researcher, ) writing_task = Task( description="""基于研究报告，撰写一篇面向开发者的 DeepSeek Agent 开发入门文章。 要求： 1. 语言通俗易懂，适合有一定 Python 基础的开发者 2. 包含具体的代码示例和架构图描述 3. 突出 DeepSeek 的独特优势 4. 篇幅控制在 800-1000 字""", expected_output="一篇完整的、可直接发布的技术文章", agent=writer, ) review_task = Task( description="""审核文章的质量，检查： 1. 技术细节是否准确 2. 代码示例是否可运行 3. 逻辑是否清晰连贯 4. 语言表达是否专业 如有问题，请标注具体位置并给出修改建议。""", expected_output="审核报告，包含通过/不通过判定和具体修改建议", agent=reviewer, ) # ========== 组建 Crew 并执行 ========== crew = Crew( agents=[researcher, writer, reviewer], tasks=[research_task, writing_task, review_task], process=Process.sequential, # 顺序执行 verbose=True, ) result = crew.kickoff() print("\n=== 最终输出 ===\n", result)

多 Agent 协作最佳实践

• 每个 Agent 只负责一个明确的角色，避免职责交叉
• 任务描述要具体，包含明确的 expected_output 格式
• 研究类 Agent 可以配置搜索工具，执行类 Agent 可以配置代码工具
• 层级流程（hierarchical）适合复杂项目，但需要额外的管理者 Agent
• 设置 allow_delegation=False 防止 Agent 之间互相推卸任务

记忆类型	存储方式	生命周期	适用场景
短期记忆	消息列表（上下文窗口）	单次会话	当前对话上下文、多轮交互
长期记忆	向量数据库 / 知识图谱	跨会话持久化	用户偏好、历史知识、项目背景
工作记忆	结构化数据存储	单次任务	任务中间结果、执行状态

from langchain.memory import ConversationBufferWindowMemory, ConversationSummaryMemory from langchain_openai import ChatOpenAI llm = ChatOpenAI( model="deepseek-chat", base_url="https://api.deepseek.com/v1", api_key="sk-your-api-key", ) # 方式 1: 滑动窗口记忆 — 只保留最近 K 轮对话 window_memory = ConversationBufferWindowMemory( k=5, # 只保留最近 5 轮对话 return_messages=True, memory_key="chat_history", ) # 方式 2: 摘要记忆 — 用 LLM 自动总结历史对话 summary_memory = ConversationSummaryMemory( llm=llm, max_token_limit=500, # 摘要最多 500 token return_messages=True, memory_key="chat_history", ) # 使用记忆的 Agent from langchain.agents import create_react_agent, AgentExecutor agent = create_react_agent(llm=llm, tools=tools, prompt=prompt) agent_with_memory = AgentExecutor( agent=agent, tools=tools, memory=window_memory, # 注入记忆 verbose=True, )

from langchain_community.vectorstores import Chroma from langchain_community.embeddings import HuggingFaceEmbeddings from langchain.schema import Document import json, os, time # 初始化向量存储 embeddings = HuggingFaceEmbeddings( model_name="BAAI/bge-small-zh-v1.5", model_kwargs={"device": "cpu"}, ) vectorstore = Chroma( persist_directory="./agent_memory", embedding_function=embeddings, collection_name="long_term_memory", ) class LongTermMemory: """长期记忆管理器""" def save_memory(self, content: str, metadata: dict = None): """保存一条记忆""" doc = Document( page_content=content, metadata=metadata or {"timestamp": time.time()}, ) vectorstore.add_documents([doc]) print(f"已保存记忆: {content[:50]}...") def recall_memory(self, query: str, k: int = 3) -> str: """检索相关记忆""" docs = vectorstore.similarity_search(query, k=k) if not docs: return "无相关历史记忆" memories = [f"- {d.page_content}" for d in docs] return "\n".join(memories) def clear_memory(self): """清空所有记忆""" vectorstore.delete_collection() print("所有记忆已清空") # 使用示例 memory = LongTermMemory() memory.save_memory("用户偏好: 喜欢简洁的回答风格，不要过多的技术细节") memory.save_memory("项目背景: 正在开发一个电商推荐系统，使用 DeepSeek 作为核心引擎") relevant = memory.recall_memory("用户的回答风格偏好") print(f"相关记忆:\n{relevant}")

推荐的记忆架构组合：

• 短期记忆：使用 ConversationBufferWindowMemory（k=10），保留最近对话
• 长期记忆：使用 ChromaDB 向量存储，存储用户偏好和项目知识
• 工作记忆：使用 Python 字典，在单次任务中传递中间结果
• 记忆检索：每次对话前自动检索相关长期记忆，注入到 System Prompt 中

1. Plan（规划阶段）：LLM 分析任务，生成详细的执行计划（步骤列表）
2. Execute（执行阶段）：Agent 按计划逐步执行，每步观察结果
3. Replan（重新规划）：如果某步失败或结果不符合预期，动态调整后续计划
4. Finalize（完成）：汇总所有步骤的结果，输出最终答案

"""DeepSeek Plan-and-Execute Agent""" from openai import OpenAI import json client = OpenAI( api_key="sk-your-api-key", base_url="https://api.deepseek.com/v1", ) # ========== 规划阶段 ========== PLANNER_PROMPT = """你是一个任务规划专家。请将用户的任务分解为清晰的执行步骤。 要求： 1. 每个步骤应该是原子操作，不可再分 2. 步骤之间要有明确的依赖关系 3. 考虑可能的异常情况和备选方案 4. 以 JSON 格式输出 输出格式： { "task": "原始任务描述", "steps": [ {"id": 1, "description": "步骤描述", "depends_on": [], "expected_output": "预期输出"}, {"id": 2, "description": "步骤描述", "depends_on": [1], "expected_output": "预期输出"} ] }""" def create_plan(task: str) -> dict: """生成执行计划""" response = client.chat.completions.create( model="deepseek-chat", messages=[ {"role": "system", "content": PLANNER_PROMPT}, {"role": "user", "content": f"请为以下任务制定执行计划:\n{task}"}, ], temperature=0.1, ) content = response.choices[0].message.content # 提取 JSON if "```json" in content: content = content.split("```json")[1].split("```")[0] return json.loads(content) # ========== 执行阶段 ========== EXECUTOR_PROMPT = """你是一个任务执行专家。请执行当前步骤，并返回结果。 当前任务的完整计划： {plan} 历史执行结果： {history} 当前步骤： 步骤 {step_id}: {step_description} 请执行此步骤并返回结果。如果步骤执行失败，请说明原因。""" def execute_step(plan: dict, step: dict, history: list) -> str: """执行单个步骤""" response = client.chat.completions.create( model="deepseek-chat", messages=[ {"role": "system", "content": EXECUTOR_PROMPT.format( plan=json.dumps(plan, ensure_ascii=False, indent=2), history=json.dumps(history, ensure_ascii=False, indent=2) if history else "无", step_id=step["id"], step_description=step["description"], )}, ], temperature=0.1, ) return response.choices[0].message.content # ========== 主循环 ========== def plan_and_execute(task: str): """Plan-and-Execute 主函数""" print("=" * 60) print(f"任务: {task}") # Phase 1: 规划 print("\n[规划阶段] 正在分析任务...") plan = create_plan(task) print(f"已生成 {len(plan['steps'])} 个执行步骤:") for s in plan["steps"]: print(f" 步骤 {s['id']}: {s['description']}") # Phase 2: 执行 print("\n[执行阶段] 开始执行...") history = [] for step in plan["steps"]: print(f"\n--- 执行步骤 {step['id']}: {step['description']} ---") result = execute_step(plan, step, history) print(f"结果: {result[:200]}...") history.append({ "step_id": step["id"], "description": step["description"], "result": result, "status": "success" if "失败" not in result else "failed", }) # Phase 3: 汇总 print("\n[汇总阶段] 生成最终报告...") return history # 测试 result = plan_and_execute("分析 DeepSeek API 的价格策略，并撰写一份竞品对比报告") print(f"\n执行完成，共完成 {len(result)} 个步骤")

动态调整计划

当某步骤执行失败时，可以让 Agent 重新规划剩余步骤。在 execute_step 返回失败后，调用 create_plan 并传入已完成步骤的历史，让模型生成新的执行计划。这种自适应能力是高级 Agent 的标志。

RAG Agent 融合了两种能力：

• 知识检索：从向量数据库中检索相关文档，提供准确的上下文信息
• 工具调用：调用外部 API、执行代码、查询数据库等操作
• 混合决策：Agent 自主判断是需要检索知识库还是调用工具

from langchain_openai import ChatOpenAI from langchain.agents import create_react_agent, AgentExecutor from langchain.tools import Tool from langchain_community.vectorstores import Chroma from langchain_community.embeddings import HuggingFaceEmbeddings from langchain.prompts import PromptTemplate llm = ChatOpenAI( model="deepseek-chat", base_url="https://api.deepseek.com/v1", api_key="sk-your-api-key", temperature=0.1, ) # ========== 初始化知识库（向量检索工具） ========== embeddings = HuggingFaceEmbeddings( model_name="BAAI/bge-small-zh-v1.5", model_kwargs={"device": "cpu"}, ) vectorstore = Chroma( persist_directory="./chroma_db", embedding_function=embeddings, ) def search_knowledge_base(query: str) -> str: """在知识库中检索相关信息""" docs = vectorstore.similarity_search(query, k=3) if not docs: return "知识库中未找到相关信息" results = [] for i, doc in enumerate(docs): source = doc.metadata.get("source", "未知") results.append(f"[来源{i+1}: {source}]\n{doc.page_content[:300]}") return "\n\n---\n\n".join(results) # ========== 定义 RAG Agent 工具集 ========== rag_agent_tools = [ Tool( name="知识库检索", func=search_knowledge_base, description="在内部知识库中搜索信息。当你需要查找公司文档、产品信息、技术规范时使用此工具。输入为搜索关键词或问题。", ), Tool( name="Python执行", func=execute_python, # 复用前面的函数 description="执行 Python 代码进行数据处理或计算。输入为 Python 代码字符串。", ), ] # RAG Agent 专用 Prompt RAG_AGENT_PROMPT = """你是一个知识库问答 Agent，可以检索内部知识库并执行代码。 可用工具： {tools} 工作流程： 1. 对于知识性问题，优先使用"知识库检索"工具 2. 对于计算需求，使用"Python执行"工具 3. 回答时请引用知识库来源，格式为 [来源X] 4. 如果知识库中没有相关信息，请如实说明 Question: {input} Thought: {agent_scratchpad}""" prompt = PromptTemplate.from_template(RAG_AGENT_PROMPT) # 创建 RAG Agent rag_agent = create_react_agent(llm=llm, tools=rag_agent_tools, prompt=prompt) rag_executor = AgentExecutor( agent=rag_agent, tools=rag_agent_tools, verbose=True, handle_parsing_errors=True, max_iterations=6, ) # 测试 RAG Agent result = rag_executor.invoke({ "input": "DeepSeek API 的收费标准是什么？如果调用 100 万 token 需要多少钱？", }) print(f"\n答案: {result['output']}")

用户问题类型	Agent 决策	使用工具
"API 怎么调用？"	知识性问题，检索知识库	知识库检索
"帮我算一下 100 万 token 的费用"	先检索价格，再计算	知识库检索 + Python执行
"对比 DeepSeek 和 GPT-4"	知识性问题，检索两个模型信息	知识库检索（多次调用）

防护层	作用	实现方式
输入验证	过滤恶意输入，防止注入攻击	正则过滤 + 内容审核 API
工具权限控制	限制 Agent 可调用的工具和参数范围	白名单 + 参数校验 + 速率限制
输出过滤	过滤敏感信息，防止数据泄露	正则脱敏 + 敏感词过滤
审计日志	记录所有 Agent 操作，便于追溯	结构化日志 + 数据库存储

import re import logging from functools import wraps from datetime import datetime # ========== 1. 输入验证 ========== class InputGuard: """输入安全检查""" # 危险模式 DANGEROUS_PATTERNS = [ r'rm\s+-rf', # 删除命令 r'DROP\s+TABLE', # SQL 注入 r'__import__', # Python 危险导入 r'os\.system', # 系统命令 r'eval\s*\(', # eval 执行 r'exec\s*\(', # exec 执行 ] MAX_INPUT_LENGTH = 10000 @classmethod def validate(cls, user_input: str) -> tuple[bool, str]: """验证输入，返回 (是否安全, 原因)""" if len(user_input) > cls.MAX_INPUT_LENGTH: return False, "输入长度超过限制" for pattern in cls.DANGEROUS_PATTERNS: if re.search(pattern, user_input, re.IGNORECASE): return False, f"检测到危险模式: {pattern}" return True, "OK" # ========== 2. 工具权限控制 ========== class ToolGuard: """工具调用安全控制""" # 工具白名单 ALLOWED_TOOLS = {"search", "calculate", "get_weather"} # 每个工具的调用次数限制 tool_call_count = {} MAX_CALLS_PER_TOOL = 10 MAX_TOTAL_CALLS = 30 @classmethod def check_tool_access(cls, tool_name: str, args: dict) -> tuple[bool, str]: """检查工具调用是否被允许""" if tool_name not in cls.ALLOWED_TOOLS: return False, f"工具 '{tool_name}' 不在白名单中" cls.tool_call_count[tool_name] = cls.tool_call_count.get(tool_name, 0) + 1 if cls.tool_call_count[tool_name] > cls.MAX_CALLS_PER_TOOL: return False, f"工具 '{tool_name}' 调用次数超限" total = sum(cls.tool_call_count.values()) if total > cls.MAX_TOTAL_CALLS: return False, "总调用次数超限" return True, "OK" # ========== 3. 输出过滤 ========== class OutputGuard: """输出安全过滤""" # 敏感信息模式 SENSITIVE_PATTERNS = [ (re.compile(r'\b\d{15,19}\b'), '[银行卡号已隐藏]'), (re.compile(r'\b1[3-9]\d{9}\b'), '[手机号已隐藏]'), (re.compile(r'\b[\w.-]+@[\w.-]+\.\w+\b'), '[邮箱已隐藏]'), (re.compile(r'\b\d{6}(?:19|20)\d{2}(?:0[1-9]|1[0-2])(?:0[1-9]|[12]\d|3[01])\d{3}[\dXx]\b'), '[身份证号已隐藏]'), ] @classmethod def sanitize(cls, output: str) -> str: """脱敏输出内容""" sanitized = output for pattern, replacement in cls.SENSITIVE_PATTERNS: sanitized = pattern.sub(replacement, sanitized) return sanitized # ========== 4. 审计日志 ========== logging.basicConfig( filename='agent_audit.log', level=logging.INFO, format='%(asctime)s | %(levelname)s | %(message)s', ) def log_agent_action(action: str, detail: dict): """记录 Agent 操作日志""" logging.info(f"{action} | {json.dumps(detail, ensure_ascii=False)}") # ========== 安全 Agent 包装器 ========== class SafeAgent: """带安全护栏的 Agent 包装器""" def __init__(self, agent_executor): self.executor = agent_executor def run(self, user_input: str) -> str: # 1. 输入验证 safe, reason = InputGuard.validate(user_input) if not safe: log_agent_action("INPUT_REJECTED", {"reason": reason}) return f"输入被拒绝: {reason}" log_agent_action("INPUT_ACCEPTED", {"input": user_input[:100]}) # 2. 执行 Agent try: result = self.executor.invoke({"input": user_input}) output = result["output"] except Exception as e: log_agent_action("EXECUTION_ERROR", {"error": str(e)}) return "Agent 执行出错，请稍后重试。" # 3. 输出过滤 safe_output = OutputGuard.sanitize(output) log_agent_action("OUTPUT_GENERATED", {"output": safe_output[:100]}) return safe_output

安全最佳实践

• 永远不要给 Agent 直接访问文件系统或数据库的权限，使用受限的 API 替代
• 代码执行工具必须使用沙箱环境（Docker 容器或 restricted Python）
• 所有外部 API 调用必须经过代理，限制速率和访问范围
• 定期审计 Agent 日志，及时发现异常行为
• 使用 System Prompt 明确告知 Agent 的安全边界和禁止行为

# pip install fastapi uvicorn pydantic from fastapi import FastAPI, HTTPException from fastapi.middleware.cors import CORSMiddleware from pydantic import BaseModel, Field from typing import Optional import uvicorn import time app = FastAPI( title="DeepSeek Agent API", version="1.0.0", description="DeepSeek Agent 智能体服务", ) # CORS 配置 app.add_middleware( CORSMiddleware, allow_origins=["*"], allow_methods=["*"], allow_headers=["*"], ) # ========== 数据模型 ========== class AgentRequest(BaseModel): query: str = Field(..., min_length=1, max_length=5000) session_id: Optional[str] = None max_steps: int = Field(default=5, ge=1, le=20) class AgentResponse(BaseModel): success: bool answer: str steps: int tools_used: list[str] execution_time: float session_id: Optional[str] # ========== API 端点 ========== @app.get("/health") async def health_check(): return {"status": "healthy", "model": "deepseek-chat"} @app.post("/agent/run", response_model=AgentResponse) async def run_agent(request: AgentRequest): """执行 Agent 任务""" start_time = time.time() tools_used = [] try: # 这里调用你的 Agent 执行器 # result = safe_agent.run(request.query) # 示例返回 answer = f"Agent 已处理查询: {request.query}" return AgentResponse( success=True, answer=answer, steps=3, tools_used=tools_used, execution_time=round(time.time() - start_time, 3), session_id=request.session_id, ) except Exception as e: raise HTTPException(status_code=500, detail=str(e)) @app.post("/agent/chat") async def chat(request: AgentRequest): """流式对话接口（SSE）""" from fastapi.responses import StreamingResponse import asyncio async def generate(): # 模拟流式输出 chunks = ["正在分析问题...", "正在调用工具...", "正在生成答案..."] for chunk in chunks: yield f"data: {chunk}\n\n" await asyncio.sleep(0.5) yield "data: [DONE]\n\n" return StreamingResponse(generate(), media_type="text/event-stream") # 启动服务 if __name__ == "__main__": uvicorn.run(app, host="0.0.0.0", port=8000, reload=True)

# Dockerfile FROM python:3.11-slim WORKDIR /app # 安装依赖 COPY requirements.txt . RUN pip install --no-cache-dir -r requirements.txt # 复制应用代码 COPY . . # 暴露端口 EXPOSE 8000 # 启动服务 CMD ["uvicorn", "agent_server:app", "--host", "0.0.0.0", "--port", "8000"] # requirements.txt # fastapi==0.115.0 # uvicorn[standard]==0.30.0 # openai==1.50.0 # langchain==0.3.0 # langchain-openai==0.2.0 # chromadb==0.5.0 # sentence-transformers==3.0.0 # pydantic==2.9.0 # docker-compose.yml # version: '3.8' # services: # agent: # build: . # ports: # - "8000:8000" # environment: # - DEEPSEEK_API_KEY=${DEEPSEEK_API_KEY} # volumes: # - ./chroma_db:/app/chroma_db # - ./agent_audit.log:/app/agent_audit.log # restart: unless-stopped # 构建和启动 # docker-compose up -d --build

# pip install prometheus-client from prometheus_client import Counter, Histogram, Gauge, generate_latest from fastapi import Response # 定义监控指标 agent_requests = Counter( 'agent_requests_total', 'Agent 请求总数', ['status'], ) agent_duration = Histogram( 'agent_request_duration_seconds', 'Agent 请求耗时', buckets=[0.5, 1, 2, 5, 10, 30], ) tool_calls = Counter( 'agent_tool_calls_total', '工具调用次数', ['tool_name'], ) active_sessions = Gauge( 'agent_active_sessions', '活跃会话数', ) # 暴露 Prometheus 指标端点 @app.get("/metrics") async def metrics(): return Response(content=generate_latest(), media_type="text/plain")

类别	检查项	工具/方案
安全	输入验证、输出过滤、工具权限	SafeAgent 包装器
监控	请求量、延迟、错误率、工具调用	Prometheus + Grafana
日志	请求日志、工具调用日志、错误日志	ELK / Loki / 结构化日志
限流	API 限流、并发控制	slowapi / Redis + 令牌桶
容错	重试机制、降级策略、健康检查	tenacity + /health 端点
部署	容器化、滚动更新、自动扩缩	Docker + K8s + HPA

部署建议

对于小规模应用，单机 Docker Compose 部署即可满足需求。对于生产级应用，建议使用 K8s 集群部署，配合 Prometheus + Grafana 监控，ELK 日志收集。Agent 服务是无状态的，可以水平扩展，但需要注意 LLM API 的并发限制。