基于 LangGraph 构建模块化 Skills 型 AI Agent

四、核心代码实现

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 搜索，计算问题调用计算器，普通问题直接回答，完全符合预期。

六、常见问题解决

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

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，实现从「问答机器人」到「智能任务执行者」的升级。