基于LangGraph的自定义SQL Agent工具执行流程构建与集成咨询

1. 定义并构建工具执行序列 #

LangGraph通过节点（Node）、条件边（Conditional Edge）和普通边实现流程控制。核心步骤是先定义每个工具对应的节点函数，再通过条件函数指定分支逻辑，最后组装成状态图。

步骤1：定义状态结构 #

用TypedDict统一存储流程中的中间结果和状态，方便节点间传递数据：

from typing import TypedDict, List
from langgraph.graph import StateGraph, END

class AgentState(TypedDict):
    list_tables_result: List[str]  # 存储list_tables的输出
    get_table_info_result: dict     # 存储get_table_info的输出
    filter_word_result: str         # 存储filter_word的输出
    execute_query_result: str       # 存储execute_query的输出
    user_query: str                 # 原始用户查询

步骤2：实现工具节点函数 #

每个节点对应一个工具调用，接收当前状态并返回更新后的状态：

# 模拟你的自定义工具函数
def list_tables():
    return ["users", "orders", "products"]

def get_table_info(table_names: List[str]):
    return {
        "users": {"columns": ["id", "name", "email"]},
        "orders": {"columns": ["order_id", "user_id", "amount"]}
    }

def filter_word(content: str, keywords: List[str]):
    return " ".join([word for word in content.split() if word not in keywords])

def execute_query(sql: str):
    return "查询结果：10条记录"

def graph_tool(final_result: str):
    return f"最终处理完成：{final_result}"

# LangGraph节点函数
def list_tables_node(state: AgentState) -> AgentState:
    result = list_tables()
    return {"list_tables_result": result, **state}

def get_table_info_node(state: AgentState) -> AgentState:
    # 后续可通过LLM选择目标表，此处模拟选择所有表
    result = get_table_info(state["list_tables_result"])
    return {"get_table_info_result": result, **state}

def filter_word_node(state: AgentState) -> AgentState:
    # 后续可通过LLM生成过滤关键词，此处模拟过滤"id"字段
    result = filter_word(str(state["get_table_info_result"]), ["id"])
    return {"filter_word_result": result, **state}

def execute_query_node(state: AgentState) -> AgentState:
    # 后续可通过LLM生成SQL，此处模拟固定查询
    result = execute_query("SELECT * FROM users LIMIT 10")
    return {"execute_query_result": result, **state}

def graph_tool_node(state: AgentState) -> AgentState:
    result = graph_tool(state["execute_query_result"])
    return {"final_result": result, **state}

步骤3：定义分支条件函数 #

条件函数接收当前状态，返回下一个节点的名称或END：

from langchain_openai import OpenAI
from langchain.prompts import PromptTemplate

llm = OpenAI(temperature=0)

# 步骤1结束后的分支判断
def after_list_tables(state: AgentState) -> str:
    prompt = PromptTemplate(
        input_variables=["tool_output", "user_query"],
        template="""用户查询：{user_query}
工具输出（可用表）：{tool_output}
如果需要获取表结构信息来生成查询，请返回"get_table_info"；如果无需继续，请返回"end"。
仅返回指定字符串，不要其他内容。"""
    )
    decision = llm.invoke(prompt.format(
        tool_output=state["list_tables_result"],
        user_query=state["user_query"]
    ))
    return decision.strip()

# 步骤2结束后的分支判断
def after_get_table_info(state: AgentState) -> str:
    prompt = PromptTemplate(
        input_variables=["tool_output", "user_query"],
        template="""用户查询：{user_query}
工具输出（表结构）：{tool_output}
如果需要过滤无关字段请返回"filter_word"；如果可以直接生成查询请返回"execute_query"；如果无需继续请返回"end"。
仅返回指定字符串，不要其他内容。"""
    )
    decision = llm.invoke(prompt.format(
        tool_output=state["get_table_info_result"],
        user_query=state["user_query"]
    ))
    return decision.strip()

# 步骤4结束后的分支判断
def after_execute_query(state: AgentState) -> str:
    prompt = PromptTemplate(
        input_variables=["tool_output", "user_query"],
        template="""用户查询：{user_query}
工具输出（查询结果）：{tool_output}
如果需要最终处理请返回"graph_tool"；如果无需继续请返回"end"。
仅返回指定字符串，不要其他内容。"""
    )
    decision = llm.invoke(prompt.format(
        tool_output=state["execute_query_result"],
        user_query=state["user_query"]
    ))
    return decision.strip()

步骤4：组装状态图 #

# 创建状态图
graph_builder = StateGraph(AgentState)

# 添加所有节点
graph_builder.add_node("list_tables", list_tables_node)
graph_builder.add_node("get_table_info", get_table_info_node)
graph_builder.add_node("filter_word", filter_word_node)
graph_builder.add_node("execute_query", execute_query_node)
graph_builder.add_node("graph_tool", graph_tool_node)

# 设置起始节点
graph_builder.set_entry_point("list_tables")

# 添加条件分支边
graph_builder.add_conditional_edges(
    "list_tables",
    after_list_tables,
    {"get_table_info": "get_table_info", "end": END}
)

graph_builder.add_conditional_edges(
    "get_table_info",
    after_get_table_info,
    {"filter_word": "filter_word", "execute_query": "execute_query", "end": END}
)

# 添加固定跳转边（步骤3完成后必须到步骤4）
graph_builder.add_edge("filter_word", "execute_query")

graph_builder.add_conditional_edges(
    "execute_query",
    after_execute_query,
    {"graph_tool": "graph_tool", "end": END}
)

# 设置graph_tool完成后结束流程
graph_builder.add_edge("graph_tool", END)

# 编译图
graph = graph_builder.compile()

运行测试 #

# 初始化状态
initial_state = {"user_query": "查询所有用户的订单信息"}

# 运行流程
result = graph.invoke(initial_state)
print(result["final_result"])

2. 有效集成LLM解析节点间输入 #

LLM的核心作用是决策分支走向和生成工具调用参数，关键是标准化提示和输出格式：

分支决策标准化 #

• 给LLM明确的指令，要求返回固定的节点名称或end，避免歧义
• 使用零样本或少样本提示，让LLM理解判断逻辑
• 对LLM输出做清洗（如strip()），防止多余字符影响分支判断

工具参数生成 #

比如在get_table_info_node中，根据用户查询和list_tables结果选择目标表：

def get_table_info_node(state: AgentState) -> AgentState:
    prompt = PromptTemplate(
        input_variables=["tables", "user_query"],
        template="""用户查询：{user_query}
可用表：{tables}
请返回需要获取结构的表名，用逗号分隔，比如"users,orders"。
仅返回表名字符串，不要其他内容。"""
    )
    table_names_str = llm.invoke(prompt.format(
        tables=state["list_tables_result"],
        user_query=state["user_query"]
    ))
    table_names = [name.strip() for name in table_names_str.split(",")]
    result = get_table_info(table_names)
    return {"get_table_info_result": result, **state}

状态传递优化 #

• 所有中间结果都存储在AgentState中，LLM可直接读取历史信息
• 在提示中包含必要上下文（如用户查询、之前的工具输出），让LLM做出准确判断

3. 最佳实践与参考思路 #

核心最佳实践 #

• 单一职责节点：每个节点只负责一个工具调用，逻辑清晰，便于调试和修改
• 结构化状态管理：用TypedDict定义状态，避免零散变量传递，提升可维护性
• 错误处理：在节点函数中添加异常捕获，处理工具调用失败的情况：

  def execute_query_node(state: AgentState) -> AgentState:
      try:
          result = execute_query("SELECT * FROM users LIMIT 10")
          return {"execute_query_result": result, **state}
      except Exception as e:
          return {"error": str(e), **state}
  

• 分支测试：单独测试每个分支路径，确保所有可能的走向都能正常执行
• LLM输出验证：对LLM返回的决策结果做合法性校验（如检查是否在允许的节点列表中），防止流程出错

参考示例思路 #

LangGraph官方SQL Agent示例的核心逻辑与你的需求一致：通过工具调用获取表信息、用LLM生成SQL并执行、根据结果决定后续操作。你可以参考其状态管理和分支控制的实现方式，核心是用条件边实现动态流程，用LLM做决策引擎。

内容的提问来源于stack exchange，提问作者Louis