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

优质文章学习记录

— 12 min read

基于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状态，跟踪执行全流程数据classAgentState(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基类，统一接口规范classBaseSkill(BaseTool):"""Skill基类，子类必须实现_run方法"""@abstractmethoddef_run(self,*args,**kwargs):"""Skill核心执行逻辑，子类必须实现"""passasyncdef_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联网搜索SkillclassSerperSearchSkill(BaseSkill): name:str="serper_search" description:str="用于联网搜索实时信息、未知知识，输入搜索关键词，返回结构化搜索结果"# 定义工具输入参数（Pydantic模型，供模型识别参数格式）classArgsSchema(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("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，支持基础加减乘除运算classCalculatorSkill(BaseSkill): name:str="calculator" description:str="用于数学计算，输入表达式字符串，返回计算结果，如'100+200*3'"# 定义工具输入参数classArgsSchema(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},{})returnf"计算结果：{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：模型决策节点，生成回复或工具调用指令defllm_node(state: AgentState)-> AgentState:# 构建模型输入：历史消息+当前用户问题 messages = state["messages"]+[HumanMessage(content=state["query"])]# 调用绑定Skill的模型 response = llm_with_skills.invoke(messages)# 更新状态：添加模型回复return{"messages":[response]}# 节点2：Skill执行节点，解析模型指令并调用对应Skilldefskill_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"]# 匹配并执行对应Skillfor 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：结果整合节点，生成最终自然语言回答defanswer_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)# 条件判断函数：模型是否需要调用Skilldefshould_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调用流程，支持传入问题并返回完整结果：

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

Read more

OpenClaw上身机器人，AI不仅能帮订外卖，还能替你跑腿了！

OpenClaw上身机器人，AI不仅能帮订外卖，还能替你跑腿了！

手把手教你一键部署OpenClaw（Clawdbot），2分钟搞定！ 过去这些年，AI大多时候还只是待在屏幕里，帮人写写字、画画图或者跑个自动脚本。但最近 OpenClaw 生态彻底爆火，两个基于它的开源项目直接打破了虚拟与现实的界限。这消息一传出来，全球搞机器人和AI的极客们都坐不住了。 就在2月23号旧金山举行的 SF OpenClaw 黑客松上，ROSClaw 项目拿下了冠军。Irvin 团队搞出了一个中间连接层，把现在最火的开源 AI Agent 平台 OpenClaw 直接插到了真实的机器人硬件上。刚拿完奖，团队就大方地宣布把项目开源了。 手把手教你一键部署OpenClaw（Clawdbot），2分钟搞定！ 手把手教你一键部署OpenClaw（Clawdbot），2分钟搞定！ 具体是怎么做到的呢？他们通过智能插件把 OpenClaw 接到了机器人操作系统（ROS 2）上，还利用 WebRTC 技术实现了超低延迟的安全连接。这意味着你在地球任何一个角落，都能远程操控那些兼容 ROS 的机器人。AI

OpenClaw安装和接入飞书机器人完整教程

OpenClaw安装和接入飞书机器人分三大部分组织回答： 1）先讲环境准备和OpenClaw基础安装（分阿里云和本地Windows两种场景）； 2）再讲飞书机器人配置（包括应用创建、通道添加、事件订阅）； 3）最后讲验证和配置AI模型。 为了更直观，在部署方式对比、配置项说明等地方用表格呈现。 这是一份完整的OpenClaw安装及接入飞书机器人的教程。将涵盖从环境准备、OpenClaw部署（含阿里云服务器和本地Windows两种方式）、AI模型（以阿里云百炼为例）配置，到最终在飞书开放平台创建并接入机器人的全流程。 第一部分：准备工作与核心认知 在开始动手前，我们需要先了解 OpenClaw 是什么，并准备好必要的账号和工具。 1.1 什么是 OpenClaw？ OpenClaw（昵称“小龙虾”，曾用名 ClawdBot / Moltbot）是一个开源的个人AI智能体框架。它本身不具备推理能力，需要对接大语言模型（如阿里云百炼、七牛云、OpenAI等）的API。它的核心价值在于： * 真正的执行能力：能通过“技能”

终极指南：10分钟搭建智能家居音乐播放系统

终极指南：10分钟搭建智能家居音乐播放系统 【免费下载链接】xiaomusic使用小爱同学播放音乐，音乐使用 yt-dlp 下载。 项目地址: https://gitcode.com/GitHub_Trending/xia/xiaomusic 还在为小爱音箱的音乐播放限制而烦恼吗？想听的歌曲搜不到，设备切换不够流畅，本地音乐管理杂乱无章？今天我们将为您介绍一个革命性的解决方案，让您的音乐体验从此无界。 痛点分析：音乐播放的三大难题 现代智能家居音乐播放面临诸多挑战：首先是设备兼容性问题，不同品牌的小爱音箱对音乐格式和播放源的支持各不相同；其次是资源整合困难，本地音乐、在线资源、个人收藏难以统一管理；最后是操作体验割裂，手机APP、语音控制、网页界面各自为政。 解决方案概述：Docker化音乐播放系统 小米音乐Docker镜像正是为解决这些痛点而生。通过容器化部署，您可以在任何支持Docker的环境中快速搭建完整的音乐播放系统，实现真正的跨设备音乐体验。 快速入门：最简化部署步骤 环境要求检查 在开始部署前，请确保您的系统满足以下基础要求： * Docke
GCC编译(6)静态库工具AR

GCC编译(6)静态库工具AR

GCC编译(6)静态库工具AR Author: Once Day Date: 2026年2月20日 一位热衷于Linux学习和开发的菜鸟，试图谱写一场冒险之旅，也许终点只是一场白日梦… 漫漫长路，有人对你微笑过嘛… 全系列文章可参考专栏: 编译构建工具链_Once-Day的博客-ZEEKLOG博客 参考文章:ar(1) - Linux manual page【Linux】ar命令：用于创建、修改和提取静态库（archive）-ZEEKLOG博客Linux命令学习手册-ar - 知乎Linux ar命令介绍 和常用示例 - Link_Z - 博客园 文章目录 * GCC编译(6)静态库工具AR * 1. AR工具概述 * 1.1 背景介绍 * 1.2 基础使用