通过上节课我们可以看到在没有告诉大模型数据库结构的情况下大模型会请求三次调用执行sql的tool来抓表、抓表结构、抓数据而且是我们通过对话大模型告诉我的本节课我们实操一下Langgraph与状态机在项目中来监控。简单来说langgraph状态机就是state、node、edge之间的流转工作State状态一个全局的字典类似Python的dict所有节点都能读写它是整个流程的“共享白板”。Node节点执行具体任务的函数比如调用大模型、查数据库接收State修改后返回新State。Edge边决定“下一个执行谁”负责把节点串起来。State 存数据Node 改数据Edge 看数据决定下一步去哪。1.我们先修改一下 sql_tool ,添加上 TABLE_SCHEMA 减少llm调用次数节省token# file tools/sql_tool.py # 导入 psycopg2用于连接和操作 PostgreSQL 数据库 import psycopg2 # 从 psycopg2 中导入 OperationalError用于捕获数据库操作异常 from psycopg2 import OperationalError # 从 pydantic 导入 BaseModel 和 Field用于定义工具输入参数的结构和描述 from pydantic import BaseModel, Field # 从 langchain.tools 导入 tool 装饰器用于将函数注册为 LangChain 工具 from langchain.tools import tool # 从 core.config 导入 Config获取数据库连接配置主机、端口、库名、用户名、密码等 from core.config import Config # 定义固定的表结构信息实际生产中可以动态从数据库读取这里先硬编码 TABLE_SCHEMA 数据库 public 模式下有一张名为 sales_data 的表结构如下 - id (INTEGER, 主键, 自增) - product_name (VARCHAR(100), 产品名称) - region (VARCHAR(50), 销售区域) - amount (DECIMAL(10,2), 销售金额) - sale_date (DATE, 销售日期) # 定义工具输入参数的 Pydantic 模型用于校验和描述 class SQLInput(BaseModel): # query 字段为字符串类型必须提供描述为只读 SQL 查询强制要求以 SELECT 开头 query: str Field( descriptionf合法的只读 SQL 查询语句必须以 SELECT 开头不要包含 INSERT/UPDATE/DELETE。\n\n当前表结构如下\n{TABLE_SCHEMA} ) # 使用 tool 装饰器将此函数声明为 LangChain 工具指定参数模型和工具描述 tool(args_schemaSQLInput, descriptionf查询 PostgreSQL 数据库中的结构化数据如财务表、销售记录。仅支持 SELECT。。表结构如下\n{TABLE_SCHEMA}) def query_sql_database(query: str) - str: 连接远程 PostgreSQL192.168.0.31执行查询返回 Markdown 表格。 # 去除查询字符串首尾的空白字符 clean_query query.strip() # 检查是否以 SELECT 开头不区分大小写否则拒绝执行并返回错误信息 if not clean_query.upper().startswith(SELECT): return ❌ 错误出于安全考虑仅支持 SELECT 只读查询。 # 构造数据库连接参数字典从 Config 中读取配置 conn_params { host: Config.PG_HOST, # 数据库主机地址 port: Config.PG_PORT, # 数据库端口 dbname: Config.PG_DB, # 数据库名称 user: Config.PG_USER, # 数据库用户名 password: Config.PG_PASSWORD, # 数据库密码 connect_timeout: 5, # 连接超时秒数 client_encoding: utf8, # 客户端字符编码防止中文乱码 } try: # 使用参数建立数据库连接 conn psycopg2.connect(**conn_params) ** 是 Python 中的字典解包Dictionary Unpacking语法也叫关键字参数解包。 简单来说** 会把一个字典拆解成 keyvalue 形式的关键字参数传递给函数。 # 设置会话为只读模式并禁用自动提交确保只读 conn.set_session(readonlyTrue, autocommitFalse) # 创建游标对象用于执行 SQL cur conn.cursor() # 执行清理后的 SQL 查询 cur.execute(clean_query) # 如果游标没有描述信息即无结果集则关闭游标和连接返回执行成功无数据 if cur.description is None: cur.close() conn.close() return ✅ 查询执行成功无返回数据。 # 从游标描述中提取列名作为表头 headers [desc[0] for desc in cur.description] # 获取所有查询结果行 rows cur.fetchall() # 定义最大返回行数防止结果过大 MAX_ROWS 30 truncated False # 如果结果行数超过最大行数则截断并标记为已截断 if len(rows) MAX_ROWS: rows rows[:MAX_ROWS] truncated True # 如果截断后没有行返回执行成功但结果为空 if not rows: return ✅ 查询执行成功但结果为空0 行。 # 构造 Markdown 表格输出 md_lines [] # 添加表头行用管道符分隔各列 md_lines.append(| | .join(headers) |) # 添加分隔行每列使用三个短横线 md_lines.append(| | .join([---] * len(headers)) |) # 遍历每一行数据将每个单元格转换为字符串若为 None 则置为空字符串 for row in rows: formatted [str(cell) if cell is not None else for cell in row] md_lines.append(| | .join(formatted) |) # 将列表合并为多行字符串 result_md \n.join(md_lines) # 如果结果被截断添加警告信息 if truncated: result_md f\n\n⚠️ 结果超过 {MAX_ROWS} 行仅展示前 {MAX_ROWS} 行。 # 关闭游标和连接释放资源 cur.close() conn.close() # 返回 Markdown 表格结果 return result_md # 捕获数据库连接或操作错误OperationalError except OperationalError as e: # 返回友好的错误提示包含主机、端口以及排查建议 return ( f❌ 数据库连接失败{Config.PG_HOST}:{Config.PG_PORT}。\n f请检查\n f1. Docker 容器是否运行\n f2. 端口映射是否正确 (-p 5432:5432)\n f3. 用户名/密码是否正确\n f错误详情: {e} ) # 捕获其他所有异常返回通用错误信息 except Exception as e: return f❌ SQL 执行异常: {e}2.新建agent/graph.py文件# file: agent/graph.py # ----- 导入部分外部依赖 ----- import json # 用于格式化打印 JSON 参数 from itertools import count # 创建无限递增计数器用于步骤编号 from langgraph.graph import StateGraph, END, MessagesState # StateGraph : 状态机图的构建器三大组件容器 # END : 内置终点标记用于边指向结束 # MessagesState: 内置状态类包含 messages: list即三大组件中的“状态” from langgraph.checkpoint.memory import MemorySaver # MemorySaver : 状态持久化插件保存每步状态快照支持跨轮记忆 from langgraph.prebuilt import tools_condition # tools_condition: 预置条件边函数判断最后一条消息是否含 tool_calls from langchain_core.messages import SystemMessage, AIMessage, HumanMessage, ToolMessage # 四种消息类型系统指令 / AI回复 / 用户输入 / 工具返回 from langchain_openai import ChatOpenAI # LLM 客户端 # 导入你的两个业务工具 from tools.rag_tool import query_knowledge_base from tools.sql_tool import query_sql_database # 全局步骤计数器与状态机无关纯日志辅助 _step_counter count(1) # 无限序列: 1,2,3,... def reset_step_counter(): 每次新问题时重置计数器 global _step_counter _step_counter count(1) def next_step() - int: 获取下一步编号 return next(_step_counter) # 核心函数构建状态机图 def build_agent_graph(llm: ChatOpenAI): 返回一个编译好的 LangGraph 状态机。 包含三大组件 - State: MessagesState消息列表 - Nodes: agent_node思考 tools_node执行 - Edges: 条件边判断是否调工具 固定边工具→Agent # ----- 1. 系统提示词固定内容注入到每次 Agent 思考 ----- SYSTEM_PROMPT ( 你是企业文档智能助手 DocMind V2.0。\n 1. 当用户询问 PDF/Word/Excel 里的内容时使用 query_knowledge_base。\n 2. 当用户询问数据报表、财务数字、统计指标时根据问题生成 SQL使用 query_sql_database 查询。\n 3. 回答要注明数据来源数据库或文档。 ) # ----- 2. 准备工具列表并绑定到 LLM ----- tools [query_knowledge_base, query_sql_database] llm_with_tools llm.bind_tools(tools) # 使 LLM 能输出 tool_calls tool_map {tool.name: tool for tool in tools} # 名字→工具对象用于执行 # # NODE 1: Agent 节点“大脑” # 功能读取整个 State调用 LLM决定下一步动作 # 输入state (MessagesState) # 输出{messages: [新的 AIMessage]}会被追加到原 messages 末尾 # def agent_node(state: MessagesState): messages state[messages] # 读取当前状态中的完整消息列表 step next_step() # 获取当前步骤编号 # --- 打印步骤标题 --- print(\n █ * 70) print(f█ 步骤 {step} │ Agent 思考阶段) print(█ * 70) # --- 显示最后一条消息便于观察输入 --- if messages: last_msg messages[-1] if isinstance(last_msg, HumanMessage): preview last_msg.content[:200].replace(\n, ) print(f 用户输入: {preview}) elif isinstance(last_msg, ToolMessage): preview last_msg.content[:300].replace(\n, ) print(f 工具返回结果预览:\n{preview}...) # --- 注入 System Prompt如果消息列表为空或首条不是系统消息 --- # 注意这里修改了本地变量 messages但并未修改原始 state if not messages or messages[0].type ! system: messages [SystemMessage(contentSYSTEM_PROMPT)] messages # --- 调用 LLM --- print(⏳ 正在调用大模型思考...) response llm_with_tools.invoke(messages) # response 是 AIMessage # --- 判断返回结果中是否包含 tool_calls --- if hasattr(response, tool_calls) and response.tool_calls: # 情况A需要调用工具 print(f✅ 决策: 需要调用 {len(response.tool_calls)} 个工具) for idx, tc in enumerate(response.tool_calls, 1): tool_name tc.get(name) tool_args tc.get(args, {}) print(f 工具 #{idx}: {tool_name}) if tool_name query_sql_database and query in tool_args: print(f SQL 语句:\n{json.dumps(tool_args[query], ensure_asciiFalse, indent4)}) else: print(f 参数: {json.dumps(tool_args, ensure_asciiFalse, indent4)}) print(f⏩ 下一步: 跳转到步骤 {step 1}工具执行) else: # 情况B直接回答无工具调用 full_content response.content print(f✅ 决策: 直接回答无需调用工具) print(f 生成回答:\n{full_content}) print(f⏩ 下一步: 结束流程) # ----- 返回值追加新的 AIMessage 到状态 ----- # 这是节点对状态的唯一修改方式通过 Reducer 追加 return {messages: [response]} # # NODE 2: 工具执行节点“手脚” # 功能读取 State 的最后一条消息必须含 tool_calls执行工具 # 输入state (MessagesState) # 输出{messages: [ToolMessage, ...]}追加到状态末尾 # def tool_executor_node(state: MessagesState): messages state[messages] last_msg messages[-1] # 取最后一条消息 tool_calls last_msg.tool_calls # 提取工具调用列表 step next_step() print(\n ▒ * 70) print(f▒ 步骤 {step} │ 工具执行阶段) print(▒ * 70) output_messages [] for idx, tc in enumerate(tool_calls, 1): tool_name tc[name] tool_args tc[args] tool_call_id tc[id] # 必须保留用于匹配工具结果 print(f▶️ 执行工具 #{idx}: {tool_name}) if tool_name query_sql_database and query in tool_args: print(f 输入 SQL:\n{json.dumps(tool_args[query], ensure_asciiFalse, indent4)}) else: print(f 输入参数: {json.dumps(tool_args, ensure_asciiFalse, indent4)}) # 执行工具从 tool_map 中获取函数 if tool_name in tool_map: try: result tool_map[tool_name].invoke(tool_args) display_result result[:800] ... if len(result) 800 else result print(f 返回结果 (预览):\n{display_result}) if len(result) 800: print(f ⚠️ 结果总长度 {len(result)} 字符已截断显示) except Exception as e: result f❌ 工具执行异常: {str(e)} print(f ❌ 执行失败: {e}) else: result f❌ 未找到工具: {tool_name} print(f ❌ 工具不存在: {tool_name}) # 将工具结果封装为 ToolMessage output_messages.append(ToolMessage(contentresult, tool_call_idtool_call_id)) print(f\n⏩ 下一步: 跳转到步骤 {step 1}回到 Agent 继续思考) print(▒ * 70) # ----- 返回值追加所有 ToolMessage ----- return {messages: output_messages} # # 构建状态机图三大组件组装 # # --- 1. 创建图实例并指定状态类型State 组件 --- # MessagesState 是 LangGraph 内置状态包含 messages: list graph StateGraph(MessagesState) # --- 2. 添加节点Node 组件 --- graph.add_node(agent, agent_node) # 节点名 agent绑订 agent_node 函数 graph.add_node(tools, tool_executor_node) # 节点名 tools绑订执行函数 # --- 3. 设置入口点从哪个节点开始执行 --- graph.set_entry_point(agent) # --- 4. 添加条件边Edge 组件 --- # tools_condition 是内置条件函数它会读取 state 的最后一条消息 # - 如果是 AIMessage 且含 tool_calls → 返回 tools # - 否则 → 返回 __end__ # 第三个参数是路由映射返回 tools 则跳转至 tools 节点 # 返回 __end__ 则结束。 graph.add_conditional_edges( agent, # 起点节点 tools_condition, # 条件函数 { tools: tools, # 若返回 tools → 去 tools 节点 __end__: END, # 若返回 __end__ → 结束 } ) # --- 5. 添加固定边Edge 组件 --- # 工具执行完毕后无条件回到 agent 节点形成 ReAct 循环 graph.add_edge(tools, agent) # --- 6. 创建内存记忆状态持久化 --- # MemorySaver 会在每个节点执行后自动保存状态快照 # 以 thread_id 索引下次调用时自动恢复。 memory MemorySaver() # --- 7. 编译图生成可执行实例 --- # compile 会验证图结构完整性并注入 checkpointer return graph.compile(checkpointermemory)3.创建一个独立的 main-react.py 文件来调用graph构建的agent与之前的main区分开# file: main-react.py DocMind V2.0 - LangGraph 白盒 ReAct 版本 完全独立于原有 main.py可并行运行互不影响 from langchain_core.messages import AIMessage, HumanMessage from langchain_openai import ChatOpenAI from core.config import Config from core.http_client import LoggingHttpClient from stores.index_pipeline import index_pipeline # 导入 LangGraph 构建器替代原来的 create_agent from agent.graph import build_agent_graph # ---------- 初始化 ---------- Config.validate() http_client LoggingHttpClient(timeout60.0) llm ChatOpenAI( modelConfig.LLM_MODEL, api_keyConfig.DEEPSEEK_API_KEY, base_urlConfig.DEEPSEEK_BASE_URL, temperature0.7, # http_clienthttp_client 之后代码会摒弃通过http日志查看链路 ) # ---------- 启动时索引与原来完全一致 ---------- print( * 60) print(DocMind V2.0 (LangGraph ReAct 版) 启动中...) print( * 60) # index_pipeline(docs_dir./docs, llmllm) 测试text to sql 为主先不加载知识库 # ---------- 创建 LangGraph Agent白盒状态机 ---------- # 注意这里不再需要传入 tools 和 system_prompt它们已经在 agent/graph.py 内部定义了 agent build_agent_graph(llm) # ---------- 对话循环 ---------- print(\n * 60) print(DocMind V2.0 (LangGraph) 已就绪) print( 核心特性: 白盒 ReAct 状态机 | 可观测的思考-行动循环) print(输入 exit 退出) print( * 60) thread_id user_001 while True: user_input input(\n你: ) if user_input.lower() exit: break if not user_input.strip(): continue # 关键变化使用 HumanMessage 包装用户输入LangGraph 标准格式 result agent.invoke( {messages: [HumanMessage(contentuser_input)]}, config{configurable: {thread_id: thread_id}} ) # 提取最后一条 AI 消息与原来逻辑完全一致 last_ai_msg None for msg in reversed(result[messages]): if isinstance(msg, AIMessage) and msg.content: last_ai_msg msg break print(- * 60) if last_ai_msg: print(f DocMind (LangGraph): {last_ai_msg.content}) else: print( DocMind (LangGraph): (未获取到有效回复)) print(- * 60)4.运行main展示结果你: 查询最新三条销售记录███████████████████████████████████████████████████████████████████████ 步骤 1 │ Agent 思考阶段██████████████████████████████████████████████████████████████████████ 用户输入: 查询最新三条销售记录⏳ 正在调用大模型思考...✅ 决策: 需要调用 1 个工具 工具 #1: query_sql_database SQL 语句:SELECT id, product_name, region, amount, sale_date FROM sales_data ORDER BY sale_date DESC LIMIT 3⏩ 下一步: 跳转到步骤 2工具执行▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒ 步骤 2 │ 工具执行阶段▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▶️ 执行工具 #1: query_sql_database 输入 SQL:SELECT id, product_name, region, amount, sale_date FROM sales_data ORDER BY sale_date DESC LIMIT 3 返回结果 (预览):| id | product_name | region | amount | sale_date || --- | --- | --- | --- | --- || 10 | 数据中台 | 华东 | 1600.00 | 2026-03-28 || 9 | 数据中台 | 华北 | 900.00 | 2026-03-15 || 3 | 智能助手 | 华东 | 1800.00 | 2026-03-10 |⏩ 下一步: 跳转到步骤 3回到 Agent 继续思考▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒███████████████████████████████████████████████████████████████████████ 步骤 3 │ Agent 思考阶段██████████████████████████████████████████████████████████████████████ 工具返回结果预览:| id | product_name | region | amount | sale_date | | --- | --- | --- | --- | --- | | 10 | 数据中台 | 华东 | 1600.00 | 2026-03-28 | | 9 | 数据中台 | 华北 | 900.00 | 2026-03-15 | | 3 | 智能助手 | 华东 | 1800.00 | 2026-03-10 |...⏳ 正在调用大模型思考...✅ 决策: 直接回答无需调用工具 生成回答:以下是数据库中最新的 **3 条销售记录** **1. 数据中台** — 华东区域- 金额**1,600.00 元**- 日期**2026-03-28** **2. 数据中台** — 华北区域- 金额**900.00 元**- 日期**2026-03-15** **3. 智能助手** — 华东区域- 金额**1,800.00 元**- 日期**2026-03-10** **数据来源**数据库 sales_data 表⏩ 下一步: 结束流程------------------------------------------------------------ DocMind (LangGraph): 以下是数据库中最新的 **3 条销售记录** **1. 数据中台** — 华东区域- 金额**1,600.00 元**- 日期**2026-03-28** **2. 数据中台** — 华北区域- 金额**900.00 元**- 日期**2026-03-15** **3. 智能助手** — 华东区域- 金额**1,800.00 元**- 日期**2026-03-10** **数据来源**数据库 sales_data 表------------------------------------------------------------你: 销售额最高的是哪条███████████████████████████████████████████████████████████████████████ 步骤 4 │ Agent 思考阶段██████████████████████████████████████████████████████████████████████ 用户输入: 销售额最高的是哪条⏳ 正在调用大模型思考...✅ 决策: 需要调用 1 个工具 工具 #1: query_sql_database SQL 语句:SELECT id, product_name, region, amount, sale_date FROM sales_data ORDER BY amount DESC LIMIT 1⏩ 下一步: 跳转到步骤 5工具执行▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒ 步骤 5 │ 工具执行阶段▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▶️ 执行工具 #1: query_sql_database 输入 SQL:SELECT id, product_name, region, amount, sale_date FROM sales_data ORDER BY amount DESC LIMIT 1 返回结果 (预览):| id | product_name | region | amount | sale_date || --- | --- | --- | --- | --- || 3 | 智能助手 | 华东 | 1800.00 | 2026-03-10 |⏩ 下一步: 跳转到步骤 6回到 Agent 继续思考▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒███████████████████████████████████████████████████████████████████████ 步骤 6 │ Agent 思考阶段██████████████████████████████████████████████████████████████████████ 工具返回结果预览:| id | product_name | region | amount | sale_date | | --- | --- | --- | --- | --- | | 3 | 智能助手 | 华东 | 1800.00 | 2026-03-10 |...⏳ 正在调用大模型思考...✅ 决策: 直接回答无需调用工具 生成回答:销售额最高的是以下这条记录 | 产品名称 | 销售区域 | 销售金额 | 销售日期 || :--- | :--- | :--- | :--- || **智能助手** | **华东** | **1,800.00 元** ✅ | 2026-03-10 | **数据来源**数据库 sales_data 表⏩ 下一步: 结束流程------------------------------------------------------------ DocMind (LangGraph): 销售额最高的是以下这条记录 | 产品名称 | 销售区域 | 销售金额 | 销售日期 || :--- | :--- | :--- | :--- || **智能助手** | **华东** | **1,800.00 元** ✅ | 2026-03-10 | **数据来源**数据库 sales_data 表------------------------------------------------------------你: 刚才那三条中销售额第二高的是哪条███████████████████████████████████████████████████████████████████████ 步骤 7 │ Agent 思考阶段██████████████████████████████████████████████████████████████████████ 用户输入: 刚才那三条中销售额第二高的是哪条⏳ 正在调用大模型思考...✅ 决策: 直接回答无需调用工具 生成回答:刚才那三条销售记录按金额从高到低排序依次是| 排名 | 产品名称 | 区域 | 金额 | 日期 || :---: | :--- | :--- | :--- | :--- || | 智能助手 | 华东 | **1,800.00 元** | 2026-03-10 || | **数据中台** | **华东** | **1,600.00 元** ✅ | **2026-03-28** || | 数据中台 | 华北 | 900.00 元 | 2026-03-15 |所以销售额 **第二高** 的是 **数据中台华东区域**金额为 **1,600.00 元**。 **数据来源**数据库 sales_data 表⏩ 下一步: 结束流程------------------------------------------------------------ DocMind (LangGraph): 刚才那三条销售记录按金额从高到低排序依次是| 排名 | 产品名称 | 区域 | 金额 | 日期 || :---: | :--- | :--- | :--- | :--- || | 智能助手 | 华东 | **1,800.00 元** | 2026-03-10 || | **数据中台** | **华东** | **1,600.00 元** ✅ | **2026-03-28** || | 数据中台 | 华北 | 900.00 元 | 2026-03-15 |所以销售额 **第二高** 的是 **数据中台华东区域**金额为 **1,600.00 元**。 **数据来源**数据库 sales_data 表------------------------------------------------------------可以看到我们通过langgraph的帮助已经可以看到智能体的执行过程了但langgraph的强大远不止于此后续我们还会使用的更多。
DocMind-python-agent-V2：Langgraph与状态机
通过上节课我们可以看到在没有告诉大模型数据库结构的情况下大模型会请求三次调用执行sql的tool来抓表、抓表结构、抓数据而且是我们通过对话大模型告诉我的本节课我们实操一下Langgraph与状态机在项目中来监控。简单来说langgraph状态机就是state、node、edge之间的流转工作State状态一个全局的字典类似Python的dict所有节点都能读写它是整个流程的“共享白板”。Node节点执行具体任务的函数比如调用大模型、查数据库接收State修改后返回新State。Edge边决定“下一个执行谁”负责把节点串起来。State 存数据Node 改数据Edge 看数据决定下一步去哪。1.我们先修改一下 sql_tool ,添加上 TABLE_SCHEMA 减少llm调用次数节省token# file tools/sql_tool.py # 导入 psycopg2用于连接和操作 PostgreSQL 数据库 import psycopg2 # 从 psycopg2 中导入 OperationalError用于捕获数据库操作异常 from psycopg2 import OperationalError # 从 pydantic 导入 BaseModel 和 Field用于定义工具输入参数的结构和描述 from pydantic import BaseModel, Field # 从 langchain.tools 导入 tool 装饰器用于将函数注册为 LangChain 工具 from langchain.tools import tool # 从 core.config 导入 Config获取数据库连接配置主机、端口、库名、用户名、密码等 from core.config import Config # 定义固定的表结构信息实际生产中可以动态从数据库读取这里先硬编码 TABLE_SCHEMA 数据库 public 模式下有一张名为 sales_data 的表结构如下 - id (INTEGER, 主键, 自增) - product_name (VARCHAR(100), 产品名称) - region (VARCHAR(50), 销售区域) - amount (DECIMAL(10,2), 销售金额) - sale_date (DATE, 销售日期) # 定义工具输入参数的 Pydantic 模型用于校验和描述 class SQLInput(BaseModel): # query 字段为字符串类型必须提供描述为只读 SQL 查询强制要求以 SELECT 开头 query: str Field( descriptionf合法的只读 SQL 查询语句必须以 SELECT 开头不要包含 INSERT/UPDATE/DELETE。\n\n当前表结构如下\n{TABLE_SCHEMA} ) # 使用 tool 装饰器将此函数声明为 LangChain 工具指定参数模型和工具描述 tool(args_schemaSQLInput, descriptionf查询 PostgreSQL 数据库中的结构化数据如财务表、销售记录。仅支持 SELECT。。表结构如下\n{TABLE_SCHEMA}) def query_sql_database(query: str) - str: 连接远程 PostgreSQL192.168.0.31执行查询返回 Markdown 表格。 # 去除查询字符串首尾的空白字符 clean_query query.strip() # 检查是否以 SELECT 开头不区分大小写否则拒绝执行并返回错误信息 if not clean_query.upper().startswith(SELECT): return ❌ 错误出于安全考虑仅支持 SELECT 只读查询。 # 构造数据库连接参数字典从 Config 中读取配置 conn_params { host: Config.PG_HOST, # 数据库主机地址 port: Config.PG_PORT, # 数据库端口 dbname: Config.PG_DB, # 数据库名称 user: Config.PG_USER, # 数据库用户名 password: Config.PG_PASSWORD, # 数据库密码 connect_timeout: 5, # 连接超时秒数 client_encoding: utf8, # 客户端字符编码防止中文乱码 } try: # 使用参数建立数据库连接 conn psycopg2.connect(**conn_params) ** 是 Python 中的字典解包Dictionary Unpacking语法也叫关键字参数解包。 简单来说** 会把一个字典拆解成 keyvalue 形式的关键字参数传递给函数。 # 设置会话为只读模式并禁用自动提交确保只读 conn.set_session(readonlyTrue, autocommitFalse) # 创建游标对象用于执行 SQL cur conn.cursor() # 执行清理后的 SQL 查询 cur.execute(clean_query) # 如果游标没有描述信息即无结果集则关闭游标和连接返回执行成功无数据 if cur.description is None: cur.close() conn.close() return ✅ 查询执行成功无返回数据。 # 从游标描述中提取列名作为表头 headers [desc[0] for desc in cur.description] # 获取所有查询结果行 rows cur.fetchall() # 定义最大返回行数防止结果过大 MAX_ROWS 30 truncated False # 如果结果行数超过最大行数则截断并标记为已截断 if len(rows) MAX_ROWS: rows rows[:MAX_ROWS] truncated True # 如果截断后没有行返回执行成功但结果为空 if not rows: return ✅ 查询执行成功但结果为空0 行。 # 构造 Markdown 表格输出 md_lines [] # 添加表头行用管道符分隔各列 md_lines.append(| | .join(headers) |) # 添加分隔行每列使用三个短横线 md_lines.append(| | .join([---] * len(headers)) |) # 遍历每一行数据将每个单元格转换为字符串若为 None 则置为空字符串 for row in rows: formatted [str(cell) if cell is not None else for cell in row] md_lines.append(| | .join(formatted) |) # 将列表合并为多行字符串 result_md \n.join(md_lines) # 如果结果被截断添加警告信息 if truncated: result_md f\n\n⚠️ 结果超过 {MAX_ROWS} 行仅展示前 {MAX_ROWS} 行。 # 关闭游标和连接释放资源 cur.close() conn.close() # 返回 Markdown 表格结果 return result_md # 捕获数据库连接或操作错误OperationalError except OperationalError as e: # 返回友好的错误提示包含主机、端口以及排查建议 return ( f❌ 数据库连接失败{Config.PG_HOST}:{Config.PG_PORT}。\n f请检查\n f1. Docker 容器是否运行\n f2. 端口映射是否正确 (-p 5432:5432)\n f3. 用户名/密码是否正确\n f错误详情: {e} ) # 捕获其他所有异常返回通用错误信息 except Exception as e: return f❌ SQL 执行异常: {e}2.新建agent/graph.py文件# file: agent/graph.py # ----- 导入部分外部依赖 ----- import json # 用于格式化打印 JSON 参数 from itertools import count # 创建无限递增计数器用于步骤编号 from langgraph.graph import StateGraph, END, MessagesState # StateGraph : 状态机图的构建器三大组件容器 # END : 内置终点标记用于边指向结束 # MessagesState: 内置状态类包含 messages: list即三大组件中的“状态” from langgraph.checkpoint.memory import MemorySaver # MemorySaver : 状态持久化插件保存每步状态快照支持跨轮记忆 from langgraph.prebuilt import tools_condition # tools_condition: 预置条件边函数判断最后一条消息是否含 tool_calls from langchain_core.messages import SystemMessage, AIMessage, HumanMessage, ToolMessage # 四种消息类型系统指令 / AI回复 / 用户输入 / 工具返回 from langchain_openai import ChatOpenAI # LLM 客户端 # 导入你的两个业务工具 from tools.rag_tool import query_knowledge_base from tools.sql_tool import query_sql_database # 全局步骤计数器与状态机无关纯日志辅助 _step_counter count(1) # 无限序列: 1,2,3,... def reset_step_counter(): 每次新问题时重置计数器 global _step_counter _step_counter count(1) def next_step() - int: 获取下一步编号 return next(_step_counter) # 核心函数构建状态机图 def build_agent_graph(llm: ChatOpenAI): 返回一个编译好的 LangGraph 状态机。 包含三大组件 - State: MessagesState消息列表 - Nodes: agent_node思考 tools_node执行 - Edges: 条件边判断是否调工具 固定边工具→Agent # ----- 1. 系统提示词固定内容注入到每次 Agent 思考 ----- SYSTEM_PROMPT ( 你是企业文档智能助手 DocMind V2.0。\n 1. 当用户询问 PDF/Word/Excel 里的内容时使用 query_knowledge_base。\n 2. 当用户询问数据报表、财务数字、统计指标时根据问题生成 SQL使用 query_sql_database 查询。\n 3. 回答要注明数据来源数据库或文档。 ) # ----- 2. 准备工具列表并绑定到 LLM ----- tools [query_knowledge_base, query_sql_database] llm_with_tools llm.bind_tools(tools) # 使 LLM 能输出 tool_calls tool_map {tool.name: tool for tool in tools} # 名字→工具对象用于执行 # # NODE 1: Agent 节点“大脑” # 功能读取整个 State调用 LLM决定下一步动作 # 输入state (MessagesState) # 输出{messages: [新的 AIMessage]}会被追加到原 messages 末尾 # def agent_node(state: MessagesState): messages state[messages] # 读取当前状态中的完整消息列表 step next_step() # 获取当前步骤编号 # --- 打印步骤标题 --- print(\n █ * 70) print(f█ 步骤 {step} │ Agent 思考阶段) print(█ * 70) # --- 显示最后一条消息便于观察输入 --- if messages: last_msg messages[-1] if isinstance(last_msg, HumanMessage): preview last_msg.content[:200].replace(\n, ) print(f 用户输入: {preview}) elif isinstance(last_msg, ToolMessage): preview last_msg.content[:300].replace(\n, ) print(f 工具返回结果预览:\n{preview}...) # --- 注入 System Prompt如果消息列表为空或首条不是系统消息 --- # 注意这里修改了本地变量 messages但并未修改原始 state if not messages or messages[0].type ! system: messages [SystemMessage(contentSYSTEM_PROMPT)] messages # --- 调用 LLM --- print(⏳ 正在调用大模型思考...) response llm_with_tools.invoke(messages) # response 是 AIMessage # --- 判断返回结果中是否包含 tool_calls --- if hasattr(response, tool_calls) and response.tool_calls: # 情况A需要调用工具 print(f✅ 决策: 需要调用 {len(response.tool_calls)} 个工具) for idx, tc in enumerate(response.tool_calls, 1): tool_name tc.get(name) tool_args tc.get(args, {}) print(f 工具 #{idx}: {tool_name}) if tool_name query_sql_database and query in tool_args: print(f SQL 语句:\n{json.dumps(tool_args[query], ensure_asciiFalse, indent4)}) else: print(f 参数: {json.dumps(tool_args, ensure_asciiFalse, indent4)}) print(f⏩ 下一步: 跳转到步骤 {step 1}工具执行) else: # 情况B直接回答无工具调用 full_content response.content print(f✅ 决策: 直接回答无需调用工具) print(f 生成回答:\n{full_content}) print(f⏩ 下一步: 结束流程) # ----- 返回值追加新的 AIMessage 到状态 ----- # 这是节点对状态的唯一修改方式通过 Reducer 追加 return {messages: [response]} # # NODE 2: 工具执行节点“手脚” # 功能读取 State 的最后一条消息必须含 tool_calls执行工具 # 输入state (MessagesState) # 输出{messages: [ToolMessage, ...]}追加到状态末尾 # def tool_executor_node(state: MessagesState): messages state[messages] last_msg messages[-1] # 取最后一条消息 tool_calls last_msg.tool_calls # 提取工具调用列表 step next_step() print(\n ▒ * 70) print(f▒ 步骤 {step} │ 工具执行阶段) print(▒ * 70) output_messages [] for idx, tc in enumerate(tool_calls, 1): tool_name tc[name] tool_args tc[args] tool_call_id tc[id] # 必须保留用于匹配工具结果 print(f▶️ 执行工具 #{idx}: {tool_name}) if tool_name query_sql_database and query in tool_args: print(f 输入 SQL:\n{json.dumps(tool_args[query], ensure_asciiFalse, indent4)}) else: print(f 输入参数: {json.dumps(tool_args, ensure_asciiFalse, indent4)}) # 执行工具从 tool_map 中获取函数 if tool_name in tool_map: try: result tool_map[tool_name].invoke(tool_args) display_result result[:800] ... if len(result) 800 else result print(f 返回结果 (预览):\n{display_result}) if len(result) 800: print(f ⚠️ 结果总长度 {len(result)} 字符已截断显示) except Exception as e: result f❌ 工具执行异常: {str(e)} print(f ❌ 执行失败: {e}) else: result f❌ 未找到工具: {tool_name} print(f ❌ 工具不存在: {tool_name}) # 将工具结果封装为 ToolMessage output_messages.append(ToolMessage(contentresult, tool_call_idtool_call_id)) print(f\n⏩ 下一步: 跳转到步骤 {step 1}回到 Agent 继续思考) print(▒ * 70) # ----- 返回值追加所有 ToolMessage ----- return {messages: output_messages} # # 构建状态机图三大组件组装 # # --- 1. 创建图实例并指定状态类型State 组件 --- # MessagesState 是 LangGraph 内置状态包含 messages: list graph StateGraph(MessagesState) # --- 2. 添加节点Node 组件 --- graph.add_node(agent, agent_node) # 节点名 agent绑订 agent_node 函数 graph.add_node(tools, tool_executor_node) # 节点名 tools绑订执行函数 # --- 3. 设置入口点从哪个节点开始执行 --- graph.set_entry_point(agent) # --- 4. 添加条件边Edge 组件 --- # tools_condition 是内置条件函数它会读取 state 的最后一条消息 # - 如果是 AIMessage 且含 tool_calls → 返回 tools # - 否则 → 返回 __end__ # 第三个参数是路由映射返回 tools 则跳转至 tools 节点 # 返回 __end__ 则结束。 graph.add_conditional_edges( agent, # 起点节点 tools_condition, # 条件函数 { tools: tools, # 若返回 tools → 去 tools 节点 __end__: END, # 若返回 __end__ → 结束 } ) # --- 5. 添加固定边Edge 组件 --- # 工具执行完毕后无条件回到 agent 节点形成 ReAct 循环 graph.add_edge(tools, agent) # --- 6. 创建内存记忆状态持久化 --- # MemorySaver 会在每个节点执行后自动保存状态快照 # 以 thread_id 索引下次调用时自动恢复。 memory MemorySaver() # --- 7. 编译图生成可执行实例 --- # compile 会验证图结构完整性并注入 checkpointer return graph.compile(checkpointermemory)3.创建一个独立的 main-react.py 文件来调用graph构建的agent与之前的main区分开# file: main-react.py DocMind V2.0 - LangGraph 白盒 ReAct 版本 完全独立于原有 main.py可并行运行互不影响 from langchain_core.messages import AIMessage, HumanMessage from langchain_openai import ChatOpenAI from core.config import Config from core.http_client import LoggingHttpClient from stores.index_pipeline import index_pipeline # 导入 LangGraph 构建器替代原来的 create_agent from agent.graph import build_agent_graph # ---------- 初始化 ---------- Config.validate() http_client LoggingHttpClient(timeout60.0) llm ChatOpenAI( modelConfig.LLM_MODEL, api_keyConfig.DEEPSEEK_API_KEY, base_urlConfig.DEEPSEEK_BASE_URL, temperature0.7, # http_clienthttp_client 之后代码会摒弃通过http日志查看链路 ) # ---------- 启动时索引与原来完全一致 ---------- print( * 60) print(DocMind V2.0 (LangGraph ReAct 版) 启动中...) print( * 60) # index_pipeline(docs_dir./docs, llmllm) 测试text to sql 为主先不加载知识库 # ---------- 创建 LangGraph Agent白盒状态机 ---------- # 注意这里不再需要传入 tools 和 system_prompt它们已经在 agent/graph.py 内部定义了 agent build_agent_graph(llm) # ---------- 对话循环 ---------- print(\n * 60) print(DocMind V2.0 (LangGraph) 已就绪) print( 核心特性: 白盒 ReAct 状态机 | 可观测的思考-行动循环) print(输入 exit 退出) print( * 60) thread_id user_001 while True: user_input input(\n你: ) if user_input.lower() exit: break if not user_input.strip(): continue # 关键变化使用 HumanMessage 包装用户输入LangGraph 标准格式 result agent.invoke( {messages: [HumanMessage(contentuser_input)]}, config{configurable: {thread_id: thread_id}} ) # 提取最后一条 AI 消息与原来逻辑完全一致 last_ai_msg None for msg in reversed(result[messages]): if isinstance(msg, AIMessage) and msg.content: last_ai_msg msg break print(- * 60) if last_ai_msg: print(f DocMind (LangGraph): {last_ai_msg.content}) else: print( DocMind (LangGraph): (未获取到有效回复)) print(- * 60)4.运行main展示结果你: 查询最新三条销售记录███████████████████████████████████████████████████████████████████████ 步骤 1 │ Agent 思考阶段██████████████████████████████████████████████████████████████████████ 用户输入: 查询最新三条销售记录⏳ 正在调用大模型思考...✅ 决策: 需要调用 1 个工具 工具 #1: query_sql_database SQL 语句:SELECT id, product_name, region, amount, sale_date FROM sales_data ORDER BY sale_date DESC LIMIT 3⏩ 下一步: 跳转到步骤 2工具执行▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒ 步骤 2 │ 工具执行阶段▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▶️ 执行工具 #1: query_sql_database 输入 SQL:SELECT id, product_name, region, amount, sale_date FROM sales_data ORDER BY sale_date DESC LIMIT 3 返回结果 (预览):| id | product_name | region | amount | sale_date || --- | --- | --- | --- | --- || 10 | 数据中台 | 华东 | 1600.00 | 2026-03-28 || 9 | 数据中台 | 华北 | 900.00 | 2026-03-15 || 3 | 智能助手 | 华东 | 1800.00 | 2026-03-10 |⏩ 下一步: 跳转到步骤 3回到 Agent 继续思考▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒███████████████████████████████████████████████████████████████████████ 步骤 3 │ Agent 思考阶段██████████████████████████████████████████████████████████████████████ 工具返回结果预览:| id | product_name | region | amount | sale_date | | --- | --- | --- | --- | --- | | 10 | 数据中台 | 华东 | 1600.00 | 2026-03-28 | | 9 | 数据中台 | 华北 | 900.00 | 2026-03-15 | | 3 | 智能助手 | 华东 | 1800.00 | 2026-03-10 |...⏳ 正在调用大模型思考...✅ 决策: 直接回答无需调用工具 生成回答:以下是数据库中最新的 **3 条销售记录** **1. 数据中台** — 华东区域- 金额**1,600.00 元**- 日期**2026-03-28** **2. 数据中台** — 华北区域- 金额**900.00 元**- 日期**2026-03-15** **3. 智能助手** — 华东区域- 金额**1,800.00 元**- 日期**2026-03-10** **数据来源**数据库 sales_data 表⏩ 下一步: 结束流程------------------------------------------------------------ DocMind (LangGraph): 以下是数据库中最新的 **3 条销售记录** **1. 数据中台** — 华东区域- 金额**1,600.00 元**- 日期**2026-03-28** **2. 数据中台** — 华北区域- 金额**900.00 元**- 日期**2026-03-15** **3. 智能助手** — 华东区域- 金额**1,800.00 元**- 日期**2026-03-10** **数据来源**数据库 sales_data 表------------------------------------------------------------你: 销售额最高的是哪条███████████████████████████████████████████████████████████████████████ 步骤 4 │ Agent 思考阶段██████████████████████████████████████████████████████████████████████ 用户输入: 销售额最高的是哪条⏳ 正在调用大模型思考...✅ 决策: 需要调用 1 个工具 工具 #1: query_sql_database SQL 语句:SELECT id, product_name, region, amount, sale_date FROM sales_data ORDER BY amount DESC LIMIT 1⏩ 下一步: 跳转到步骤 5工具执行▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒ 步骤 5 │ 工具执行阶段▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▶️ 执行工具 #1: query_sql_database 输入 SQL:SELECT id, product_name, region, amount, sale_date FROM sales_data ORDER BY amount DESC LIMIT 1 返回结果 (预览):| id | product_name | region | amount | sale_date || --- | --- | --- | --- | --- || 3 | 智能助手 | 华东 | 1800.00 | 2026-03-10 |⏩ 下一步: 跳转到步骤 6回到 Agent 继续思考▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒███████████████████████████████████████████████████████████████████████ 步骤 6 │ Agent 思考阶段██████████████████████████████████████████████████████████████████████ 工具返回结果预览:| id | product_name | region | amount | sale_date | | --- | --- | --- | --- | --- | | 3 | 智能助手 | 华东 | 1800.00 | 2026-03-10 |...⏳ 正在调用大模型思考...✅ 决策: 直接回答无需调用工具 生成回答:销售额最高的是以下这条记录 | 产品名称 | 销售区域 | 销售金额 | 销售日期 || :--- | :--- | :--- | :--- || **智能助手** | **华东** | **1,800.00 元** ✅ | 2026-03-10 | **数据来源**数据库 sales_data 表⏩ 下一步: 结束流程------------------------------------------------------------ DocMind (LangGraph): 销售额最高的是以下这条记录 | 产品名称 | 销售区域 | 销售金额 | 销售日期 || :--- | :--- | :--- | :--- || **智能助手** | **华东** | **1,800.00 元** ✅ | 2026-03-10 | **数据来源**数据库 sales_data 表------------------------------------------------------------你: 刚才那三条中销售额第二高的是哪条███████████████████████████████████████████████████████████████████████ 步骤 7 │ Agent 思考阶段██████████████████████████████████████████████████████████████████████ 用户输入: 刚才那三条中销售额第二高的是哪条⏳ 正在调用大模型思考...✅ 决策: 直接回答无需调用工具 生成回答:刚才那三条销售记录按金额从高到低排序依次是| 排名 | 产品名称 | 区域 | 金额 | 日期 || :---: | :--- | :--- | :--- | :--- || | 智能助手 | 华东 | **1,800.00 元** | 2026-03-10 || | **数据中台** | **华东** | **1,600.00 元** ✅ | **2026-03-28** || | 数据中台 | 华北 | 900.00 元 | 2026-03-15 |所以销售额 **第二高** 的是 **数据中台华东区域**金额为 **1,600.00 元**。 **数据来源**数据库 sales_data 表⏩ 下一步: 结束流程------------------------------------------------------------ DocMind (LangGraph): 刚才那三条销售记录按金额从高到低排序依次是| 排名 | 产品名称 | 区域 | 金额 | 日期 || :---: | :--- | :--- | :--- | :--- || | 智能助手 | 华东 | **1,800.00 元** | 2026-03-10 || | **数据中台** | **华东** | **1,600.00 元** ✅ | **2026-03-28** || | 数据中台 | 华北 | 900.00 元 | 2026-03-15 |所以销售额 **第二高** 的是 **数据中台华东区域**金额为 **1,600.00 元**。 **数据来源**数据库 sales_data 表------------------------------------------------------------可以看到我们通过langgraph的帮助已经可以看到智能体的执行过程了但langgraph的强大远不止于此后续我们还会使用的更多。
