个人主页 for_ever_love__ 欢迎各位大佬莅临其他栏目: 大模型开发从0到1 其他栏目: iOS项目总结大全 其他栏目: 我想学python了 其他栏目: iOS UI 文章目录LangChain Text-to-SQL让不会写 SQL 的人查数据库一、先说风险二、最简实现三、关键表结构怎么写3.1 只给相关的表3.2 字段要有中文注释3.3 给出示例值四、自动提取 schema表多的时候五、用官方链create_sql_query_chain六、安全层必做七、完整链路生成 → 校验 → 执行 → 自然语言回答八、失败重试把报错告诉模型九、用 Agent 让它可以多步查询十、常见坑十一、小结LangChain Text-to-SQL让不会写 SQL 的人查数据库“上个月华东区销售额前三的产品是哪些”——业务同事每天都要问这类问题每次都得找数据分析师写 SQL。Text-to-SQL 让模型来写 SQL。但这件事风险很高这篇讲怎么做到能用且不出事。一、先说风险在讲怎么做之前必须说清楚三条红线绝不能让模型执行任意 SQLDROP / DELETE / UPDATE 都可能被生成绝不能把整个数据库暴露给它只给它需要的表必须有行数和耗时限制一条没加 LIMIT 的查询能拖垮生产库。这三条守不住就别做这个功能。二、最简实现fromlangchain_openaiimportChatOpenAIfromlangchain_core.promptsimportChatPromptTemplatefromlangchain_core.output_parsersimportStrOutputParser SCHEMA 表 orders: id, user_id, product_id, amount, status, created_at 表 products: id, name, category, price 表 users: id, name, city, vip_level promptChatPromptTemplate.from_template( 你是 SQL 专家。根据表结构把用户问题转成 SQL。 【表结构】 {schema} 【规则】 1. 只输出 SQL 语句不要解释不要 markdown 代码块 2. 只使用上面给出的表不要假设其他表存在 3. 必须加 LIMIT 100 用户问题{question} )chainprompt|ChatOpenAI(modelgpt-4o-mini,temperature0)|StrOutputParser()sqlchain.invoke({schema:SCHEMA,question:上个月销售额最高的3个产品})print(sql)# SELECT p.name, SUM(o.amount) AS total# FROM orders o JOIN products p ON o.product_id p.id# WHERE o.created_at 2024-09-01 AND o.created_at 2024-10-01# GROUP BY p.id, p.name ORDER BY total DESC LIMIT 3三、关键表结构怎么写模型只能看到你给它的 schema。写不好SQL 就写不对。3.1 只给相关的表# ❌ 把 50 张表全塞进去schemaget_all_tables_schema()# 上万 token模型还会选错表# ✅ 只给业务相关的 3~5 张SCHEMA 表 orders订单: id, user_id, product_id, amount(元), status, created_at 表 products商品: id, name, category, price 3.2 字段要有中文注释-- ❌ 模型不知道 status 里存的是什么statusTEXT-- ✅statusTEXT-- 订单状态pending待付款 / paid已付款 / shipped已发货 / done已完成枚举值必须写出来否则模型会猜status completed而实际值是done查询结果为空。3.3 给出示例值SCHEMA 表 orders: id 订单ID user_id 用户ID amount 订单金额元整数 status 状态pending/paid/shipped/done/cancelled created_at 下单时间格式 YYYY-MM-DD HH:MM:SS 示例数据 id1001, user_id88, amount299, statuspaid, created_at2024-09-15 10:30:00 给几行示例数据模型对格式的理解会准确很多——这和 Few-shot 是一个道理。四、自动提取 schema表多的时候表多了手写不现实可以从数据库读fromlangchain_community.utilitiesimportSQLDatabase dbSQLDatabase.from_uri(sqlite:///data/app.db)print(db.get_usable_table_names())# [orders, products, users]print(db.get_table_info([orders,products]))# CREATE TABLE orders (...)# SELECT * FROM orders LIMIT 3;# ...SQLDatabase会自动带上SELECT * LIMIT 3的示例数据很贴心。但自动提取的 schema 没有中文注释建议手动补一层COMMENTS{orders.status:状态pending/paid/shipped/done/cancelled,orders.amount:金额元,}schemadb.get_table_info()schema\n\n字段说明\n\n.join(f{k}--{v}fork,vinCOMMENTS.items())五、用官方链create_sql_query_chainfromlangchain_community.utilitiesimportSQLDatabasefromlangchain_openaiimportChatOpenAIfromlangchain.chainsimportcreate_sql_query_chain dbSQLDatabase.from_uri(sqlite:///data/app.db)llmChatOpenAI(modelgpt-4o-mini,temperature0)chaincreate_sql_query_chain(llm,db)sqlchain.invoke({question:上个月有多少订单})它内部帮你处理了 schema 注入和 prompt 模板。但它只生成 SQL不执行——执行要自己做这正好方便加安全层。六、安全层必做importreimportsqlite3 DANGEROUS[drop,delete,update,insert,alter,create,truncate,attach,pragma,grant]defsafe_execute(sql:str,conn,max_rows100,timeout10):# 1. 清洗ssql.strip().rstrip(;)# 去掉可能的 markdown 代码块sre.sub(r^(?:sql)?|$,,s,flagsre.M).strip()# 2. 只允许 SELECT / WITHfirst_words.split()[0].upper()iffirst_wordnotin(SELECT,WITH):returnNone,f只允许 SELECT 查询收到的是{first_word}# 3. 危险关键词检查按词边界匹配避免误伤列名lows.lower()forkwinDANGEROUS:ifre.search(rf\b{kw}\b,low):returnNone,f查询包含不允许的关键字{kw}# 4. 强制加 LIMITifnotre.search(r\blimit\b,low):sf LIMIT{max_rows}# 5. 执行带超时try:conn.execute(fPRAGMA busy_timeout {timeout*1000})curconn.execute(s)rowscur.fetchmany(max_rows)cols[d[0]fordincur.description]return{columns:cols,rows:rows},NoneexceptExceptionase:returnNone,f执行失败{e}五道关卡缺一不可。更稳妥的做法——用只读账号连接数据库dbSQLDatabase.from_uri(postgresql://readonly_user:xxxhost/db)这样即使 SQL 层被绕过数据库层还能兜住。七、完整链路生成 → 校验 → 执行 → 自然语言回答fromlangchain_core.runnablesimportRunnableLambda,RunnablePassthroughdefquery_and_answer(question:str):# 1. 生成 SQLsqlchain.invoke({question:question})# 2. 安全执行result,errsafe_execute(sql,conn)iferr:return{answer:f查询失败{err},sql:sql}# 3. 让模型把结果说成人话answer(ChatPromptTemplate.from_template( 根据查询结果回答用户问题。只陈述数据里有的事实不要推测。 用户问题{question} SQL{sql} 查询结果最多 100 行{result} 用一两句话回答关键数字要带上。数据为空就说没有查到数据。 )|llm|StrOutputParser()).invoke({question:question,sql:sql,result:str(result)})return{answer:answer,sql:sql,data:result}把 SQL 也返回给前端展示——让用户能看到系统是怎么查的既能建立信任也方便发现错误。八、失败重试把报错告诉模型模型写的 SQL 经常有语法错误或字段不存在。把数据库报错返回给它让它改FIX_PROMPTChatPromptTemplate.from_template( 下面的 SQL 执行失败了请根据错误信息修正它。 只输出修正后的 SQL不要解释。 【表结构】 {schema} 【原始问题】 {question} 【失败的 SQL】 {sql} 【错误信息】 {error} )defquery_with_retry(question:str,max_attempts3):sqlchain.invoke({question:question})foriinrange(max_attempts):result,errsafe_execute(sql,conn)iferrisNone:returnsql,resultifimax_attempts-1:returnsql,None# 让模型根据报错修正sql(FIX_PROMPT|llm|StrOutputParser()).invoke({schema:SCHEMA,question:question,sql:sql,error:err,})returnsql,None实测这一招能修回大部分语法错误和字段名错误比单纯重新生成不带错误信息成功率高得多。九、用 Agent 让它可以多步查询复杂问题可能需要先查有哪些表→再查数据fromlangchain_community.agent_toolkitsimportcreate_sql_agentfromlangchain_community.utilitiesimportSQLDatabase dbSQLDatabase.from_uri(sqlite:///data/app.db)agentcreate_sql_agent(ChatOpenAI(modelgpt-4o-mini,temperature0),dbdb,agent_typeopenai-tools,verboseTrue,)agent.invoke({input:哪个城市的 VIP 用户平均消费最高})但生产环境我更推荐自己写固定链理由Agent 会自己执行 SQL安全层不好插步数不可控成本不可控出错难排查。只有在探索性数据分析场景才用 Agent面向业务的固定查询用前面的固定链。十、常见坑现象原因解法生成的 SQL 查不到数据枚举值猜错schema 里写明枚举字段名不存在schema 过时从数据库自动生成 schema查询把库拖垮没加 LIMIT / 没超时安全层必做日期算错模型不知道上个月是哪天在 prompt 里注入当前日期数字类型混乱amount 是分还是元schema 里注明单位中文表名/字段处理引号问题schema 里给出正确的引用方式日期那条特别容易中fromdatetimeimportdatetime promptprompt.partial(todaydatetime.now().strftime(%Y-%m-%d))# prompt 里加一句今天是 {today}模型不知道今天是几号必须告诉它。十一、小结三条红线只允许 SELECT、只暴露必要的表、必须加 LIMIT 和超时schema 三要素只给相关表、字段名加中文注释和枚举值、附示例数据SQLDatabase能自动提取 schemaget_table_info但中文注释要手动补安全层五关清洗 markdown → 只允许 SELECT/WITH → 危险词检查→ 强制 LIMIT → 超时外加只读数据库账号把报错返回给模型让它修正比重新生成成功率高得多prompt 里注入当前日期否则上个月这类相对时间会算错生产推荐固定链Agent 只在探索性场景用。下一篇讲知识图谱 RAG——处理谁和谁什么关系这类问题。