基于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

高效OCR识别新选择｜DeepSeek-OCR-WEBUI本地部署指南

高效OCR识别新选择｜DeepSeek-OCR-WEBUI本地部署指南 1. 为什么你需要一个本地OCR系统？ 你有没有遇到过这样的情况：手头有一堆扫描件、发票、合同或者老照片，想要提取里面的文字，却发现复制粘贴根本不管用？传统OCR工具要么识别不准，要么不支持复杂排版，更别说手写体或模糊图像了。这时候，你就需要一个真正“聪明”的OCR系统。 而今天要介绍的 DeepSeek-OCR-WEBUI，正是这样一个能看懂图、识得字、还能说清楚内容的智能OCR解决方案。它基于国产自研的大模型技术，不仅中文识别精准，还自带可视化界面，部署后直接通过网页操作，像用手机App一样简单。 更重要的是——它是可以完全私有化部署的。你的数据不会上传到任何云端，所有处理都在本地完成，安全又高效。无论是企业文档自动化，还是个人资料数字化，都是理想选择。 2. DeepSeek-OCR-WEBUI 是什么？ 2.1 核心能力一览 DeepSeek-OCR-WEBUI 并不是一个简单的文字识别工具，而是一套完整的图像理解与文本提取系统。它的背后是 DeepSeek 团队开源的高性能 OCR 大模

WebRTC 架构概览（整体框架篇）

WebRTC 架构概览（整体框架篇） 本文是 WebRTC 系列专栏的第二篇，将深入剖析 WebRTC 的整体架构，包括浏览器中的实现架构、API 体系、信令流程以及底层媒体引擎 libwebrtc 的结构。 目录 1. WebRTC 在浏览器中的架构 2. API 体系详解 3. WebRTC 信令流程概览 4. 媒体引擎结构（libwebrtc 概览） 5. 总结 1. WebRTC 在浏览器中的架构 1.1 整体架构图 ┌─────────────────────────────────────────────────────────────────────────┐ │ Web Application │ │ (JavaScript / HTML) │ └─────────────────────────────────────────────────────────────────────────┘ │ ▼ ┌───────────────────────────────────────────────────────────────────────

Web 可访问性最佳实践：构建人人可用的前端界面

Web 可访问性最佳实践：构建人人可用的前端界面 代码如诗，包容如画。让我们用可访问性的理念，构建出人人都能使用的前端界面。 什么是 Web 可访问性？ Web 可访问性（Web Accessibility）是指网站、工具和技术能够被所有人使用，包括那些有 disabilities 的人。这意味着无论用户的能力如何，他们都应该能够感知、理解、导航和与 Web 内容交互。 为什么 Web 可访问性很重要？ 1. 法律要求：许多国家和地区都有法律法规要求网站必须具有可访问性。 2. 扩大用户群体：约 15% 的世界人口生活有某种形式的 disability，可访问性可以让更多人使用你的网站。 3. SEO 优化：搜索引擎爬虫依赖于可访问性良好的网站结构。 4. 更好的用户体验：可访问性改进通常会使所有用户受益，而不仅仅是那些有 disabilities 的用户。 5. 社会责任：

微信小程序如何优雅地跳转外部链接？WebView + 复制方案实战

在做小程序开发的过程中，我们经常会遇到这样一个需求： 👉 用户在小程序里点开一个课程/资料，需要跳转到公司内部的学习系统或者外部网站。 问题来了： * 小程序禁止直接用 <a> 标签跳转外部网页 * 也不能像浏览器里那样用 window.open * 那么，怎么实现呢？ 这篇文章我会结合实际项目，聊聊 两种常见方案： 1. 业务域名 + WebView 打开外部链接 2. 不在业务域名里的 → 自动复制链接 1️⃣ 背景：小程序的安全限制 微信对小程序的外部链接有严格限制： * 只能通过 <WebView /> 组件来加载 H5 页面。 * 这个 H5 的域名，必须提前在 小程序后台 → 开发设置 → 业务域名 配置。 * 没配置的域名，一律打不开。 所以，解决问题的第一步就是搞清楚： 👉 目标链接的域名是否可控、