基于 LangGraph 实现模块化 Skills 型 AI Agent

基于 LangGraph+DeepSeek+Serper 实现模块化 Skills 型 AI Agent

在 AI Agent 的落地实践中，模块化 Skills 设计是提升 Agent 可扩展性、可维护性的核心方案——将搜索、计算、文件处理等能力封装为独立 Skills，Agent 可根据需求自主调用，无需修改核心流程。本文将基于 LangGraph、DeepSeek 大模型和 Serper 搜索工具，手把手带你实现一个具备工具调用能力的 Skills 型 AI Agent，同时解决开发中常见的 MRO 冲突、Pydantic 验证等问题，代码可直接复制运行。

一、前言：为什么选择 Skills 型 Agent？

传统 AI Agent 多采用「硬编码工具调用」的方式，新增能力需修改核心逻辑，耦合度高且难以维护。而Skills 型 Agent将能力拆分为独立的 Skill 模块，每个 Skill 遵循统一接口，具备以下优势：

1. 模块化解耦：新增/修改 Skill 无需改动 Agent 核心流程，即插即用；
2. 智能决策：大模型自主判断是否调用 Skill、调用哪个 Skill，无需人工干预；
3. 可扩展性强：支持搜索、计算、代码解释、数据库查询等多类型 Skill 集成；
4. 流程标准化：基于 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+ 流程编排」的三层架构，核心分工如下：

1. 状态层：通过 AgentState 跟踪用户问题、历史消息、工具调用结果、最终回答，实现数据流转；
2. Skill 层：将 Serper 搜索、计算器等能力封装为独立 Skill，遵循统一接口；
3. 流程层：基于 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 年人工智能发展趋势'")
    
    # 绑定参数 schema
    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()
                })
            
             json.dumps(formatted_results, ensure_ascii=, indent=)
         Exception  e:
             ToolException()

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)'")
    
    # 绑定参数 schema
    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, : , : })
                    state[].append(ToolMessage(content=, tool_call_id=skill_id))
                
    
     {: tool_results, : state[]}


 () -> AgentState:
    
    prompt = 
    
    final_response = llm.invoke([HumanMessage(content=prompt)])
    
     {: 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 搜索，计算问题调用计算器，普通问题直接回答，完全符合预期。

六、常见问题解决

在开发过程中，我们遇到了几个典型问题，以下是解决方案总结：

1. MRO 冲突（Cannot create a consistent method resolution order）：BaseSkill 同时继承 ABC 和 BaseTool 导致，删除 ABC 继承即可，BaseTool 已内置抽象基类能力；
2. ModuleNotFoundError: No module named 'langchain_core.pydantic_v1'：LangChain 新版移除 pydantic_v1，直接从 pydantic 导入 Field；
3. Pydantic ValidationError（Field required）：将 Skill 参数定义为类属性导致，改为通过 ArgsSchema 定义输入参数，实例化时无需传参。

七、优化与扩展方向

7.1 核心优化

1. 异常处理增强：使用 tenacity 库为 API 调用添加重试机制，应对网络波动；
2. 结果筛选优化：对 Serper 搜索结果增加相关性排序，过滤低质量信息；
3. 状态持久化：集成 Redis 实现 Agent 状态持久化，支持会话恢复；
4. 多轮对话优化：保留历史消息，实现上下文感知的连续交互。

7.2 能力扩展

1. 新增 Skill：封装文件读写、代码解释、数据库查询、天气查询等能力，遵循 BaseSkill 接口即可；
2. 并行 Skill 调用：利用 LangGraph 的并行节点，实现多个 Skill 同时执行，提升效率；
3. 记忆模块：增加短期/长期记忆，让 Agent 记住用户偏好和历史对话；
4. 可视化调试：使用 agent.get_graph().draw_mermaid() 生成流程图，便于流程调试。

八、总结

本文基于 LangGraph、DeepSeek 和 Serper，实现了一个模块化的 Skills 型 AI Agent，完整覆盖了状态管理、Skill 封装、流程编排、智能工具调用四大核心环节。该方案具备以下核心优势：

• 模块化设计：Skill 独立封装，新增能力无需修改核心流程；
• 智能化调度：DeepSeek 模型自主决策工具调用，灵活适配不同场景；
• 可扩展性强：支持多类型 Skill 集成、多轮对话、并行调用等复杂场景；
• 生产级可用：解决了开发中常见的兼容性、验证问题，代码可直接落地使用。

你可以基于本文的代码框架，根据业务需求扩展更多 Skill，打造专属的 AI Agent，实现从「问答机器人」到「智能任务执行者」的升级。