AI大模型应用开发-用LangChain构建PAL应用：SQL的生成与执行

PAL（Program-Aided Language models） 思想成为大模型 Agent 领域的重要范式。核心思路是 LLM 只负责语言任务，复杂的逻辑/计算交由程序执行。
通过合理设计 prompt，模型生成代码/SQL/逻辑描述，外部程序再执行，得到结果后反馈给 LLM，LLM 再生成最终答案。
本文将通过一个 LangChain + MySQL + Postgres Checkpoint 实例，完整演示 PAL 的设计流程，帮助大家理解和复现。
项目结构

1. .
2. ├── llm_env.py         # 初始化 LLM
3. └── main.py            # PAL 交互主流程

复制代码

llm_env.py：初始化 LLM

1. from langchain.chat_models import init_chat_model
3. llm = init_chat_model("gpt-4o-mini", model_provider="openai")

复制代码

简单封装一个 LLM 对象，这里用 gpt-4o-mini，通过 llm_env.llm 调用。
main.py：PAL 主流程

导入库 & 初始化

1. import os
2. import sys
4. sys.path.append(os.getcwd())
6. from langchain_community.utilities import SQLDatabase
7. from langchain_core.prompts import ChatPromptTemplate
8. from langchain_community.tools.sql_database.tool import QuerySQLDatabaseTool
9. from langgraph.graph import START, StateGraph
10. from langgraph.checkpoint.postgres import PostgresSaver
11. import time
12. from typing import TypedDict, Annotated
13. from llm_set import llm_env
15. llm = llm_env.llm

复制代码

亮点：

• 用 StateGraph 组织整个流程
  
• 用 PostgresSaver 持久化 checkpoint，方便中断恢复
  

初始化 MySQL 数据库

1. db = SQLDatabase.from_uri(
2.     "mysql+pymysql://root:123456@localhost:3306/javademo",
3.     engine_args={"pool_size": 5, "max_overflow": 10},
4. )

复制代码

数据库连接，示例用 javademo 库，用户可根据实际修改。
State 定义

1. class State(TypedDict):
2.     """State for the demo."""
4.     question: str
5.     query: str
6.     result: str
7.     answer: str
8.     approved: bool

复制代码

定义流程中的共享变量，典型 PAL 模式的中间态。
Prompt 设计

1. system_message = """
2. Given an input question, create a syntactically correct {dialect} query to
3. run to help find the answer. Unless the user specifies in his question a
4. specific number of examples they wish to obtain, always limit your query to
5. at most {top_k} results. You can order the results by a relevant column to
6. return the most interesting examples in the database.
8. Never query for all the columns from a specific table, only ask for a the
9. few relevant columns given the question.
11. Pay attention to use only the column names that you can see in the schema
12. description. Be careful to not query for columns that do not exist. Also,
13. pay attention to which column is in which table.
15. Only use the following tables:
16. {table_info}
17. """
19. user_prompt = "Question:{input}"
21. query_prompt_template = ChatPromptTemplate(
22.     [("system", system_message), ("human", user_prompt)],
23. )

复制代码

亮点：

• 系统提示明确要求 安全、规范 的 SQL
  
• 限定 top_k 结果
  
• 避免 SELECT *
  

LLM 生成 SQL

1. class QueryOutput(TypedDict):
2.     """Generated the SQL query."""
3.     query: Annotated[str, "Syntactically valid SQL query."]
5. def write_query(state: State):
6.     """Generate SQL query to fetch information."""
7.     prompt = query_prompt_template.invoke(
8.         {
9.             "dialect": db.dialect,
10.             "top_k": 5,
11.             "table_info": db.get_table_info(),
12.             "input": state["question"],
13.         }
14.     )
16.     structured_llm = llm.with_structured_output(QueryOutput)
18.     result = structured_llm.invoke(prompt)
20.     return {"query": result["query"]}

复制代码

PAL 核心步骤 1
LLM 不直接回答问题，而是生成 SQL 查询。
用户确认

1. def wait_for_user_approve(state: State):
2.     """Pause here and wait for user approval before executing query."""
3.     try:
4.         user_approval = input("Do you want to go to execute query? (yes/no): ")
5.     except Exception:
6.         user_approval = "no"
8.     if user_approval.lower() == "yes":
9.         return {
10.             "query": state["query"],
11.             "approved": True,
12.         }
13.     else:
14.         return {
15.             "query": state["query"],
16.             "approved": False,
17.         }

复制代码

用户确认 SQL 是否执行，保证安全性 —— 很关键。
执行 SQL

1. def excute_query(state: State):
2.     """Execute the SQL query and return the result."""
3.     if state["approved"]:
4.         execute_query_tool = QuerySQLDatabaseTool(db=db)
5.         return {"result": execute_query_tool.invoke(state["query"])}
6.     else:
7.         return {"result": "excute denied."}

复制代码

PAL 核心步骤 2
SQL 执行交给程序完成，LLM 不直接操作数据库。
回答用户

1. def generate_answer(state: State):
2.     """Answer question using retrieved information as context."""
3.     if state["approved"]:
4.         prompt = (
5.             "Given the following user question, corresponding SQL query, "
6.             "and SQL result, answer the user question.\n\n"
7.             f'Question: {state["question"]}\n'
8.             f'SQL Query: {state["query"]}\n'
9.             f'SQL Result: {state["result"]}'
10.         )
11.         response = llm.invoke(prompt)
12.         return {"answer": response.content}
14.     else:
15.         prompt = f'{"同意" if state["approved"] else "拒绝"} 用户拒绝当前执行'
16.         response = llm.invoke(prompt)
17.         return {"answer": response.content}

复制代码

PAL 核心步骤 3
LLM 根据 SQL 结果生成最终自然语言答案。
流程图

1. graph_builder = StateGraph(State).add_sequence(
2.     [write_query, wait_for_user_approve, excute_query, generate_answer]
3. )
5. graph_builder.add_edge(START, "write_query")

复制代码

用 LangGraph 编排整个 PAL 流程。
持久化 checkpoint + 启动循环

1. DB_URI = "postgresql://postgres:123456@localhost:5432/langchaindemo?sslmode=disable"
2. with PostgresSaver.from_conn_string(DB_URI) as checkpointer:
3.     checkpointer.setup()
5.     input_thread_id = input("输入thread_id:")
6.     time_str = time.strftime("%Y%m%d", time.localtime())
7.     config = {"configurable": {"thread_id": f"{time_str}-{input_thread_id}-agent-demo"}}
9.     graph = graph_builder.compile(checkpointer=checkpointer)
11.     print("输入问题，输入 exit 退出。")
12.     while True:
13.         query = input("你: ")
14.         if query.strip().lower() == "exit":
15.             break
16.         response = graph.invoke(
17.             {"question": query},
18.             config,
19.         )
21.         print(response)

复制代码

• Checkpoint 存入 Postgres
  
• 用户可断点续跑
  
• CLI 交互友好
  

总结：PAL 模式的好处

1.LLM 不做逻辑执行，只负责「写程序」
2.复杂逻辑交给程序完成，结果回传给 LLM
3.SQL 查询避免安全风险
4.有明确「用户确认」步骤
5.checkpoint 持久化，支持中断恢复
运行效果示例

 

来源：程序园用户自行投稿发布，如果侵权，请联系站长删除
免责声明：如果侵犯了您的权益，请联系站长，我们会及时删除侵权内容，谢谢合作！