基于 LangGraph+DeepSeek+Serper 实现模块化 Skills 型 AI Agent
在 AI Agent 的落地实践中，模块化 Skills 设计是提升可扩展性与可维护性的核心方案——将搜索、计算、文件处理等能力封装为独立 Skill，Agent 可根据需求自主调用，无需修改核心流程。本文将基于 LangGraph、DeepSeek 大模型和 Serper 搜索工具，带你实现一个具备工具调用能力的 Skills 型 AI Agent，同时解决开发中常见的 MRO 冲突、Pydantic 验证等问题。
一、前言：为什么选择 Skills 型 Agent？
传统 AI Agent 多采用「硬编码工具调用」的方式，新增能力需修改核心逻辑，耦合度高且难以维护。而Skills 型 Agent将能力拆分为独立的 Skill 模块，每个 Skill 遵循统一接口，具备以下优势：
- 模块化解耦：新增/修改 Skill 无需改动 Agent 核心流程，即插即用；
- 智能决策：大模型自主判断是否调用 Skill、调用哪个 Skill，无需人工干预；
- 可扩展性强：支持搜索、计算、代码解释、数据库查询等多类型 Skill 集成；
- 流程标准化：基于 LangGraph 实现状态管理与流程编排，支持多轮对话、并行工具调用。
本次实现的核心技术栈：
- LangGraph：Agent 状态管理、流程编排核心框架；
- DeepSeek：提供自然语言理解、工具调用决策、结果生成能力；
- Serper：提供联网搜索能力，封装为独立 Skill；
- LangChain：提供工具封装、模型交互的基础能力。
二、环境准备
2.1 依赖安装
首先创建虚拟环境并安装核心依赖，确保版本兼容：
# 创建虚拟环境（可选）
python -m venv agent-env
# 激活环境（Windows）
agent-env\Scripts\activate
# 激活环境（Mac/Linux）
source agent-env/bin/activate
# 安装核心依赖
pip install langgraph langchain-deepseek langchain-core langchain-community python-dotenv pydantic
2.2 环境变量配置
创建 .env 文件，配置 DeepSeek 和 Serper 的 API 密钥（需在 DeepSeek 平台和 Serper 平台申请）：
# .env 文件
DEEPSEEK_API_KEY="你的 DeepSeek API Key"
SERPER_API_KEY="你的 Serper API Key"
三、核心架构设计
本次实现的 Skills 型 Agent 采用「状态管理 + 模块化 Skill+ 流程编排」的三层架构，核心分工如下：
- 状态层：通过
AgentState跟踪用户问题、历史消息、工具调用结果、最终回答，实现数据流转； - Skill 层：将 Serper 搜索、计算器等能力封装为独立 Skill，遵循统一接口；
- 流程层：基于 LangGraph 定义「模型决策→Skill 执行→结果整合」的执行流程，通过条件边实现智能调度。
四、核心代码实现
4.1 状态定义（AgentState）
LangGraph 通过状态（State）管理 Agent 执行过程中的所有数据，我们定义包含用户输入、历史消息、工具结果、最终回答的状态结构：
from typing import TypedDict, Annotated, List, Dict, Any
from langchain_core.messages import BaseMessage
import operator
# 定义 Agent 状态，跟踪执行全流程数据
class AgentState(TypedDict):
# 用户输入的问题
query: str
# 历史消息（用户问题、模型回复、工具结果）
messages: Annotated[List[BaseMessage], operator.add]
# 工具调用结果
tool_results: List[Dict[str, Any]]
# 最终生成的回答
answer: str
Annotated[List[BaseMessage], operator.add]：实现消息列表的追加合并，支持多轮对话；- 状态字段覆盖 Agent 执行全生命周期，确保数据可追溯。
4.2 Skill 基类封装
为了统一 Skill 接口，我们定义 BaseSkill 基类，所有自定义 Skill 都继承该类，强制实现 _run 方法（核心执行逻辑）。注意：LangChain 的 BaseTool 已内置抽象基类能力，无需额外继承 ABC，避免 MRO 冲突。
from abc import abstractmethod
from langchain_core.tools import BaseTool, ToolException
from pydantic import BaseModel, Field
# Skill 基类，统一接口规范
class BaseSkill(BaseTool):
"""Skill 基类，子类必须实现_run 方法"""
@abstractmethod
def _run(self, *args, **kwargs):
"""Skill 核心执行逻辑，子类必须实现"""
pass
async def _arun(self, *args, **kwargs):
"""异步执行（可选，本次实现同步逻辑）"""
raise NotImplementedError("异步执行暂未实现")
4.3 自定义 Skill 实现
基于 BaseSkill 封装两个核心 Skill：Serper 联网搜索 Skill（使用 LangChain 官方 GoogleSerperAPIWrapper 简化逻辑）和计算器 Skill，遵循 LangChain 工具规范，通过 ArgsSchema 定义输入参数（解决 Pydantic 验证报错问题）。
4.3.1 Serper 搜索 Skill
from langchain_community.utilities import GoogleSerperAPIWrapper
import json
# Serper 联网搜索 Skill
class SerperSearchSkill(BaseSkill):
name: str = "serper_search"
description: str = "用于联网搜索实时信息、未知知识，输入搜索关键词，返回结构化搜索结果"
# 定义工具输入参数（Pydantic 模型，供模型识别参数格式）
class ArgsSchema(BaseModel):
query: str = Field(description="搜索关键词，需精准明确，如'2025 年人工智能发展趋势'")
args_schema: type[BaseModel] = ArgsSchema
def _run(self, query: str) -> str:
"""执行搜索逻辑，调用官方 Wrapper 简化请求"""
try:
# 初始化 Serper 搜索（自动读取环境变量 SERPER_API_KEY）
search = GoogleSerperAPIWrapper(
gl="cn", # 搜索地区：中国
hl="zh-CN", # 搜索语言：中文
k=5 # 返回 5 条结果，避免信息过载
)
# 执行搜索并获取结果
search_results = search.results(query)
# 格式化结果（提取标题、链接、摘要）
formatted_results = []
for item in search_results.get("organic", []):
formatted_results.append({
"title": item.get("title"),
"link": item.get("link"),
"snippet": item.get("snippet")
})
# 返回 JSON 格式结果，便于模型解析
return json.dumps(formatted_results, ensure_ascii=False, indent=2)
except Exception as e:
raise ToolException(f"Serper 搜索失败：{str(e)}")
4.3.2 计算器 Skill
# 计算器 Skill，支持基础加减乘除运算
class CalculatorSkill(BaseSkill):
name: str = "calculator"
description: str = "用于数学计算，输入表达式字符串，返回计算结果，如'100+200*3'"
# 定义工具输入参数
class ArgsSchema(BaseModel):
expression: str = Field(description="数学表达式，仅支持加减乘除，如'5*(10+20)'")
args_schema: type[BaseModel] = ArgsSchema
def _run(self, expression: str) -> str:
"""执行计算逻辑，安全过滤恶意代码"""
try:
# 安全计算：禁用内置函数，仅支持基础运算
result = eval(expression, {"__builtins__": None}, {})
return f"计算结果：{result}"
except Exception as e:
raise ToolException(f"计算失败：{str(e)}")
4.3.3 Skill 实例化
# 实例化所有 Skill
search_skill = SerperSearchSkill()
calculator_skill = CalculatorSkill()
# 收集 Skill 列表，供模型绑定
skills = [search_skill, calculator_skill]
4.4 大模型初始化
初始化 DeepSeek 大模型，绑定所有 Skill，让模型感知可用工具并生成调用指令：
from langchain_deepseek import ChatDeepSeek
from dotenv import load_dotenv
import os
# 加载环境变量
load_dotenv()
# 初始化 DeepSeek 模型
llm = ChatDeepSeek(
model="deepseek-chat",
api_key=os.getenv("DEEPSEEK_API_KEY"),
temperature=0.1, # 低温度保证结果准确性
max_tokens=2048 # 最大生成长度
)
# 绑定 Skill 到模型，模型可自主调用工具
llm_with_skills = llm.bind_tools(skills)
4.5 LangGraph 流程编排
将 Agent 的执行逻辑拆分为独立节点，通过 StateGraph 连接节点，设置条件边实现「模型决策是否调用 Skill」的智能调度。
4.5.1 定义流程节点
from langgraph.graph import StateGraph, START, END
from langchain_core.messages import ToolMessage, HumanMessage
# 节点 1：模型决策节点，生成回复或工具调用指令
def llm_node(state: AgentState) -> AgentState:
# 构建模型输入：历史消息 + 当前用户问题
messages = state["messages"] + [HumanMessage(content=state["query"])]
# 调用绑定 Skill 的模型
response = llm_with_skills.invoke(messages)
# 更新状态：添加模型回复
return {"messages": [response]}
# 节点 2：Skill 执行节点，解析模型指令并调用对应 Skill
def skill_executor_node(state: AgentState) -> AgentState:
last_message = state["messages"][-1]
tool_results = []
# 遍历所有工具调用（支持多 Skill 并行调用）
for tool_call in last_message.tool_calls:
skill_name = tool_call["name"]
skill_args = tool_call["args"]
skill_id = tool_call["id"]
# 匹配并执行对应 Skill
for skill in skills:
if skill.name == skill_name:
try:
# 执行 Skill
result = skill._run(**skill_args)
tool_results.append({
"skill_name": skill_name,
"result": result,
"status": "success"
})
# 生成工具结果消息，供模型后续整合
state["messages"].append(ToolMessage(content=result, tool_call_id=skill_id))
except Exception as e:
tool_results.append({
"skill_name": skill_name,
"result": f"执行失败：{str(e)}",
"status": "failed"
})
state["messages"].append(ToolMessage(content=f"执行失败：{str(e)}", tool_call_id=skill_id))
break
# 更新状态：工具结果 + 消息列表
return {"tool_results": tool_results, "messages": state["messages"]}
# 节点 3：结果整合节点，生成最终自然语言回答
def answer_node(state: AgentState) -> AgentState:
# 构建整合提示词
prompt = f"""
基于以下信息，回答用户问题：
{state['query']}
工具调用结果：{state['tool_results']}
要求：
1. 语言自然流畅，符合中文表达习惯；
2. 准确引用工具结果中的关键信息；
3. 若工具调用失败，明确告知用户并给出替代方案。
"""
# 调用模型生成最终回答
final_response = llm.invoke([HumanMessage(content=prompt)])
# 更新状态：最终回答
return {"answer": final_response.content}
4.5.2 构建流程与条件边
# 初始化流程构建器
workflow = StateGraph(AgentState)
# 添加节点
workflow.add_node("llm_node", llm_node)
workflow.add_node("skill_executor_node", skill_executor_node)
workflow.add_node("answer_node", answer_node)
# 条件判断函数：模型是否需要调用 Skill
def should_call_skill(state: AgentState) -> str:
last_message = state["messages"][-1]
# 若模型返回工具调用指令，则执行 Skill；否则直接生成回答
return "skill_executor_node" if last_message.tool_calls else "answer_node"
# 连接节点与条件边
workflow.add_edge(START, "llm_node") # 起始→模型节点
workflow.add_conditional_edges("llm_node", should_call_skill) # 模型决策分支
workflow.add_edge("skill_executor_node", "answer_node") # Skill 执行→结果整合
workflow.add_edge("answer_node", END) # 结果整合→结束
# 编译流程，生成可执行的 Agent
agent = workflow.compile()
4.6 Agent 测试函数
封装测试函数，简化 Agent 调用流程，支持传入问题并返回完整结果：
def run_agent(query: str) -> Dict[str, Any]:
"""执行 AI Agent，返回完整结果"""
# 初始化状态
initial_state = {
"query": query,
"messages": [],
"tool_results": [],
"answer": ""
}
# 运行 Agent
result = agent.invoke(initial_state)
return result
五、场景测试
我们通过三个典型场景测试 Agent 的能力：联网搜索、数学计算、普通问答，验证 Skill 调用与流程编排的有效性。
5.1 场景 1：联网搜索（调用 Serper Skill）
if __name__ == "__main__":
# 测试 1：实时信息查询
print("=== 测试场景 1：联网搜索 ===")
result1 = run_agent("2025 年全球人工智能领域的重大突破有哪些？")
print(f"用户问题：{result1['query']}")
print(f"工具调用结果：{result1['tool_results']}")
print(f"最终回答：{result1['answer']}\n")
5.2 场景 2：数学计算（调用计算器 Skill）
# 测试 2：数学计算
print("=== 测试场景 2：数学计算 ===")
result2 = run_agent("(100 + 50) * 3 - 80")
print(f"用户问题：{result2['query']}")
print(f"工具调用结果：{result2['tool_results']}")
print(f"最终回答：{result2['answer']}\n")
5.3 场景 3：普通问答（无需调用 Skill）
# 测试 3：普通问答
print("=== 测试场景 3：普通问答 ===")
result3 = run_agent("什么是人工智能？")
print(f"用户问题：{result3['query']}")
print(f"工具调用结果：{result3['tool_results']}")
print(f"最终回答：{result3['answer']}")
运行代码后，Agent 会根据问题类型自主决策：实时问题调用 Serper 搜索，计算问题调用计算器，普通问题直接回答，完全符合预期。
六、常见问题解决
在开发过程中，我们遇到了几个典型问题，以下是解决方案总结：
- MRO 冲突（Cannot create a consistent method resolution order）：
BaseSkill同时继承ABC和BaseTool导致，删除ABC继承即可，BaseTool已内置抽象基类能力； - ModuleNotFoundError: No module named 'langchain_core.pydantic_v1'：LangChain 新版移除
pydantic_v1，直接从pydantic导入Field； - Pydantic ValidationError（Field required）：将 Skill 参数定义为类属性导致，改为通过
ArgsSchema定义输入参数，实例化时无需传参。
七、优化与扩展方向
7.1 核心优化
- 异常处理增强：使用
tenacity库为 API 调用添加重试机制，应对网络波动； - 结果筛选优化：对 Serper 搜索结果增加相关性排序，过滤低质量信息；
- 状态持久化：集成 Redis 实现 Agent 状态持久化，支持会话恢复；
- 多轮对话优化：保留历史消息，实现上下文感知的连续交互。
7.2 能力扩展
- 新增 Skill：封装文件读写、代码解释、数据库查询、天气查询等能力，遵循
BaseSkill接口即可； - 并行 Skill 调用：利用 LangGraph 的并行节点，实现多个 Skill 同时执行，提升效率；
- 记忆模块：增加短期/长期记忆，让 Agent 记住用户偏好和历史对话；
- 可视化调试：使用
agent.get_graph().draw_mermaid()生成流程图，便于流程调试。
八、总结
本文基于 LangGraph、DeepSeek 和 Serper，实现了一个模块化的 Skills 型 AI Agent，完整覆盖了状态管理、Skill 封装、流程编排、智能工具调用四大核心环节。该方案具备以下核心优势：
- 模块化设计：Skill 独立封装，新增能力无需修改核心流程；
- 智能化调度：DeepSeek 模型自主决策工具调用，灵活适配不同场景；
- 可扩展性强：支持多类型 Skill 集成、多轮对话、并行调用等复杂场景；
- 生产级可用：解决了开发中常见的兼容性、验证问题，代码可直接落地使用。
你可以基于本文的代码框架，根据业务需求扩展更多 Skill，打造专属的 AI Agent，实现从「问答机器人」到「智能任务执行者」的升级。
