[大模型架构] LangGraph AI 工作流编排（16）

一、可调用工具（Tools）的核心定义与设计原则

视频首先明确了 “可调用工具” 的本质的是 “封装外部功能（如文件操作、API 调用、模型推理）的标准化接口”，需满足 “LangGraph 节点可调用、参数可配置、结果可解析” 的核心要求，同时遵循三大设计原则：

（一）工具设计三大原则

1. 单一职责原则：一个工具仅实现一个核心功能（如 “PDF 文本提取工具” 仅负责提取文本，不包含后续清洗、分析逻辑），确保工具的复用性与维护性；
2. 参数标准化原则：输入输出参数需明确类型注解（如file_path: str“result: List[str]），支持必填 / 可选配置，避免参数歧义；
3. 容错性原则：内置异常处理逻辑（如文件不存在、API 调用超时），返回标准化错误信息，避免工具调用失败导致整个工作流中断。

（二）工具的核心组成结构

一个可被 LangGraph 调用的工具需包含三部分核心内容，示例结构如下：

python

运行

# 工具核心组成：元数据+参数定义+执行逻辑 class PDFTextExtractTool: # 1. 工具元数据（描述信息，便于前端展示与选择） metadata = { "name": "pdf_text_extract", # 工具唯一标识（英文无空格） "description": "提取PDF文件中的文本内容，支持多页PDF与扫描件（需OCR支持）", # 功能描述 "author": "AI-Workflow-Team", "version": "1.0.0" } # 2. 参数定义（支持必填/可选、类型限制、默认值） params = [ { "key": "file_path", "label": "PDF文件路径", "type": "string", "required": True, # 必填参数 "description": "本地PDF文件的绝对路径（如C:/docs/test.pdf）" }, { "key": "use_ocr", "label": 是否启用OCR, "type": "boolean", "required": False, "default": False, # 默认值 "description": "扫描件PDF需启用OCR，纯文本PDF无需启用" }, { "key": "page_range", "label": "提取页码范围", "type": "string", "required": False, "default": "all", "description": "格式：1-5（第1-5页）、single（单页）、all（全部）" } ] # 3. 执行逻辑（核心功能实现，输入参数字典，返回结果字典） def run(self, params: dict) -> dict: try: file_path = params["file_path"] use_ocr = params.get("use_ocr", self.params[1]["default"]) page_range = params.get("page_range", self.params[2]["default"]) # 核心功能逻辑（如PDF文本提取） text_content = self._extract_text(file_path, use_ocr, page_range) return { "success": True, "data": text_content, # 工具执行结果 "message": "PDF文本提取成功" } except FileNotFoundError: return {"success": False, "data": None, "message": f"文件不存在：{params['file_path']}"} except Exception as e: return {"success": False, "data": None, "message": f"提取失败：{str(e)}"} # 内部辅助方法（封装具体实现） def _extract_text(self, file_path: str, use_ocr: bool, page_range: str) -> List[str]: # 实际PDF提取逻辑（如使用PyPDF2提取纯文本，pytesseract实现OCR） pass

二、LangChain 工具适配（快速复用生态工具）

视频重点讲解了如何适配 LangChain 生态中的现成工具（如FileToolkit“RequestsToolkit），避免重复开发，同时确保其可被 LangGraph 节点调用：

（一）LangChain 工具适配步骤

1. 安装 LangChain 工具依赖：

   bash

   运行

   pip install langchain langchain-community python-dotenv # 核心依赖+环境变量管理

2. 直接复用 LangChain 内置工具：
   • 以 “文件读取工具” 为例，直接导入 LangChain 的ReadFileTool，无需自定义：

     python

     运行

     from langchain_community.tools.file_management.read import ReadFileTool from langchain_community.tools.file_management.write import WriteFileTool # 初始化LangChain工具（已满足标准化参数与执行逻辑） read_file_tool = ReadFileTool() write_file_tool = WriteFileTool()

3. 自定义 LangChain 工具包装器（适配 LangGraph 调用）：
   • LangChain 工具默认返回str或dict，需包装为 LangGraph State 兼容的格式，示例包装器：

     python

     运行

     def langchain_tool_wrapper(tool, params: dict) -> dict: """将LangChain工具包装为标准化输出""" try: # LangChain工具调用（需将参数转换为工具所需格式） result = tool.run(**params) return { "success": True, "data": result, "message": f"{tool.name}执行成功" } except Exception as e: return { "success": False, "data": None, "message": f"{tool.name}执行失败：{str(e)}" }

（二）LangChain 工具参数适配

• LangChain 工具的参数需与 LangGraph State 字段对应，例如将 “文件路径” 从 State 传递给ReadFileTool：

  python

  运行

  # LangGraph节点中调用LangChain工具 def read_file_node(state: WorkflowState) -> WorkflowState: # 从State中获取工具参数 tool_params = {"file_path": state.file_path} # 调用包装后的LangChain工具 result = langchain_tool_wrapper(read_file_tool, tool_params) # 将工具结果写入State if result["success"]: state.file_content = result["data"] else: state.error_message = result["message"] state.workflow_status = "failed" return state

三、LangGraph 工具节点集成（核心实操）

视频的核心环节是 “将自定义工具 / LangChain 工具封装为 LangGraph 节点”，实现工具调用与工作流的无缝衔接，分为 “自定义工具节点” 与 “LangChain 工具节点” 两种场景：

（一）自定义工具节点封装

以 “PDF 文本提取工具” 为例，完整实现 LangGraph 工具节点：

1. 步骤 1：定义工具类（参考第一部分结构）：
   • 实现PDFTextExtractTool类，包含元数据、参数定义、执行逻辑。
2. 步骤 2：封装为 LangGraph 节点函数：

   python

   运行

   from langgraph.graph import State # 定义工作流State（包含工具参数与结果字段） class PDFProcessState(State): # 工具输入参数 file_path: str use_ocr: bool = False # 工具输出结果 pdf_text: List[str] = [] # 工作流状态 workflow_status: str = "running" error_message: str = "" # 步骤3：创建工具节点函数（调用自定义工具） def pdf_extract_node(state: PDFProcessState) -> PDFProcessState: # 1. 初始化工具 pdf_tool = PDFTextExtractTool() # 2. 组装工具参数（从State中提取） tool_params = { "file_path": state.file_path, "use_ocr": state.use_ocr } # 3. 调用工具 result = pdf_tool.run(tool_params) # 4. 处理工具结果（更新State） if result["success"]: state.pdf_text = result["data"] state.workflow_status = "completed" else: state.error_message = result["message"] state.workflow_status = "failed" return state

（二）LangGraph 工具节点调用工作流

将工具节点集成到工作流中，实现 “工具调用 + 后续处理” 的完整逻辑：

python

运行

from langgraph import Graph # 1. 创建Graph实例 graph = Graph(PDFProcessState) # 2. 添加工具节点与后续处理节点 graph.add_node("pdf_extract", pdf_extract_node) # 工具节点 graph.add_node("text_clean", text_clean_node) # 后续文本清洗节点（自定义） # 3. 定义节点流转逻辑 graph.add_edge("pdf_extract", "text_clean") # 工具执行完成→文本清洗 graph.set_entry_point("pdf_extract") # 入口：工具节点 graph.set_finish_point("text_clean") # 结束：清洗完成 # 4. 编译并执行工作流 app = graph.compile() initial_state = PDFProcessState( file_path="C:/docs/contract.pdf", use_ocr=True # 扫描件PDF启用OCR ) final_state = app.invoke(initial_state) # 5. 查看结果 if final_state.workflow_status == "completed": print("PDF文本提取成功：", final_state.pdf_text) else: print("执行失败：", final_state.error_message)

（三）多工具串联调用示例

演示 “文件读取→文本提取→结果写入” 的多工具协同工作流：

python

运行

# 定义包含多工具参数的State class MultiToolState(State): input_file: str output_file: str extracted_text: str = "" status: str = "running" # 工具节点1：读取PDF（自定义工具） def extract_pdf_node(state: MultiToolState) -> MultiToolState: tool = PDFTextExtractTool() result = tool.run({"file_path": state.input_file}) if result["success"]: state.extracted_text = "\n".join(result["data"]) return state # 工具节点2：写入文件（LangChain工具） def write_file_node(state: MultiToolState) -> MultiToolState: result = langchain_tool_wrapper(write_file_tool, { "file_path": state.output_file, "text": state.extracted_text }) if result["success"]: state.status = "completed" else: state.status = "failed" return state # 构建多工具工作流 graph = Graph(MultiToolState) graph.add_node("extract_pdf", extract_pdf_node) graph.add_node("write_file", write_file_node) graph.add_edge("extract_pdf", "write_file") graph.set_entry_point("extract_pdf") graph.set_finish_point("write_file") # 执行 app = graph.compile() app.invoke(MultiToolState( input_file="C:/docs/input.pdf", output_file="C:/docs/output.txt" ))

四、Electron 前端交互对接（工具参数配置与结果展示）

视频最后实现了 “前端可视化配置工具参数→主进程调用 LangGraph 工具节点→前端展示执行结果” 的闭环，让非技术用户也能通过界面控制工具调用：

（一）前端工具参数配置面板

1. 工具元数据展示：前端通过 IPC 获取工具的metadata与params，动态生成配置表单（如 “PDF 文件路径” 对应文件选择器，“是否启用 OCR” 对应开关）；
2. 前端组件示例（React）：

   tsx

   import { useState } from 'react'; import { Button, Input, Switch, Form, Upload } from 'antd'; const ToolConfigPanel = ({ tool }) => { const [form] = Form.useForm(); const [loading, setLoading] = useState(false); // 提交工具参数（触发工作流） const handleSubmit = async () => { const params = await form.validateFields(); setLoading(true); // 调用IPC，传递工具ID与参数 const result = await window.electronAPI.runToolWorkflow({ toolId: tool.metadata.name, params }); setLoading(false); // 展示结果 alert(result.message); }; return ( <Form form={form} layout="vertical"> {tool.params.map((param) => ( <Form.Item key={param.key} label={param.label} rules={param.required ? [{ required: true, message: `请输入${param.label}` }] : []} > {param.type === "string" && param.key.includes("file_path") ? ( <Upload type="file" accept=".pdf"> <Button>选择文件</Button> </Upload> ) : param.type === "boolean" ? ( <Switch defaultChecked={param.default} /> ) : ( <Input placeholder={param.description} /> )} </Form.Item> ))} <Button type="primary" onClick={handleSubmit} loading={loading}> 执行工具 </Button> </Form> ); };

（二）Electron 主进程调用逻辑

1. IPC 事件处理：接收前端工具参数，调用 LangGraph 工作流：

   javascript

   运行

   // src/main/toolWorkflow.js const { ipcMain } = require('electron'); const { runPdfExtractWorkflow } = require('./langgraph/workflows'); // 监听前端工具工作流请求 ipcMain.handle('runToolWorkflow', async (_, { toolId, params }) => { try { let result; // 根据工具ID选择对应的工作流 if (toolId === "pdf_text_extract") { result = await runPdfExtractWorkflow(params); // 调用LangGraph工作流 } return result; } catch (error) { return { success: false, message: error.message }; } });

2. 工作流执行结果返回：LangGraph 工作流执行完成后，将结果（成功 / 失败、数据 / 错误信息）通过 IPC 返回前端，前端展示给用户。

五、工具开发与集成的常见问题避坑