SQL生成前有那些信息
query # 用户原始问题
table_infos # 可使用的表、字段、字段类型、字段描述、示例值
metric_infos # 可参考的业务指标、指标口径、依赖字段
date_info # 当前日期、星期、季度
db_info # 当前数据库方言和版本如果缺少这些上下文，模型也能写出一条“看起来像 SQL”的语句，但那条 SQL 很可能用错表、用错字段、写错指标口径，甚至不符合当前数据库版本。所以本章的重点不是让模型自由发挥，而是让模型在前面整理好的约束范围内生成 SQL
生成SQL-generate_sql
llm结合上述信息，生成sql.
其中最关键的约束是：输出必须只包含一条完整 SQL 语句的纯文本.需要注意的是，很多大模型在写代码时，会习惯性输出 Markdown 代码块.
但项目拿到 SQL 后要直接交给数据库校验和执行。如果 SQL 字符串里混入 Markdown 代码块，数据库并不认识这些符号，就会报语法错误
所以提示词必须明确要求:
- 不要解释
- 不要 Markdown 代码块
- 不要 ```sql
- 只输出 SQL 本身
核心代码
async def generate_sql(state: DataAgentState, runtime: Runtime[DataAgentContext]):writer = runtime.stream_writerwriter("生成SQL")# 这些上下文都由前置节点准备完成，模型只在给定表、字段、指标口径范围内生成 SQLtable_infos = state["table_infos"]metric_infos = state["metric_infos"]date_info = state["date_info"]db_info = state["db_info"]query = state["query"]prompt = PromptTemplate(template=load_prompt("generate_sql"),input_variables=["table_infos", "metric_infos", "date_info", "db_info", "query"],)# 生成 SQL 只需要一段纯文本，所以这里使用 StrOutputParseroutput_parser = StrOutputParser()chain = prompt | llm | output_parserresult = await chain.ainvoke({# YAML 更适合放进提示词：保留嵌套结构、顺序和中文说明，方便模型理解表字段关系"table_infos": yaml.dump(table_infos, allow_unicode=True, sort_keys=False),"metric_infos": yaml.dump(metric_infos, allow_unicode=True, sort_keys=False),"date_info": yaml.dump(date_info, allow_unicode=True, sort_keys=False),"db_info": yaml.dump(db_info, allow_unicode=True, sort_keys=False),"query": query,})logger.info(f"生成的SQL：{result}")return {"sql": result}两个点值得注意：
- 结构化上下文会先转成 YAML。
- yaml.dump(table_infos, allow_unicode=True, sort_keys=False)
- 输出解析器使用的是
StrOutputParser，不是JsonOutputParser- 因为这一节点只需要一条 SQL 字符串，不需要 JSON 对象
校验SQL： validate_sql
生成 SQL 之后，不能直接执行,大模型也仍然可能出错。常见错误包括：
- 表名写错；
- 字段名写错；
- 字段别名写错；
join条件不完整；- 聚合函数和
group by不匹配； - SQL 方言不符合当前数据库；
- 使用了提示词里没有提供的表或字段。
explain <generated_sql>
EXPLAIN 原本是用来看 SQL 执行计划的，但这里主要利用它的副作用：让数据库提前解析 SQL,若 SQL 里有字段不存在，数据库会直接报错。
这条错误信息非常有用。后面 correct_sql 可以把它连同原 SQL 一起交给大模型，让模型做有依据的修正
validate_sql核心代码
async def validate_sql(state: DataAgentState, runtime: Runtime[DataAgentContext]):writer = runtime.stream_writerwriter("校验SQL")# 读取 generate_sql 写入状态的 SQL。sql = state["sql"]# SQL 可用性必须交给真实数仓判断，这里从运行时 context 中取 DW Repository。dw_mysql_repository: DWMySQLRepository = runtime.context["dw_mysql_repository"]try:# validate 内部使用 explain <sql>，只关心数据库能否成功解析这条 SQL。await dw_mysql_repository.validate(sql)logger.info("SQL语法正确")return {"error": None}except Exception as e:# 不直接抛异常，而是把错误信息写入 state，交给条件边判断。logger.info(f"SQL语法错误：{str(e)}")return {"error": str(e)}校验失败时，节点没有直接抛异常，而是把错误转成字符串，写入状态，return {"error": str(e)}，这是为了交给 LangGraph 的条件分支处理
也就是说，validate_sql 只负责判断 SQL 是否有问题，不负责决定下一步去哪。下一步走 run_sql 还是 correct_sql，由图结构来决定