05. 从 Demo 到可上线的 LangChain 应用

Demo 的成功标准通常是“能得到一个像样的答案”。生产系统的成功标准完全不同：错误是否被限制、行为是否可观察、敏感数据是否安全、修改后是否会回归。

本篇用自然语言转 SQL 作为主线，因为它能清楚暴露模型应用的核心原则：模型输出永远是不可信输入。

Demo 能跑不等于可以上线

一个最小 SQL 助手可能只有两步：

sql = model.invoke(f"把这个问题转换成 SQL：{question}")
rows = database.execute(sql)

这段代码在演示中可能跑通，却同时留下严重问题：

• 模型可能生成 DELETE、DROP 或多条语句；
• 可能查询用户无权访问的表；
• 可能返回数百万行；
• 复杂查询可能长期占用数据库；
• Prompt 中的“只允许查询”不是权限控制；
• 出错时不知道是生成、校验还是执行阶段失败。

上线不是再写一个更严厉的 Prompt，而是把权限边界写进确定性程序。

模型输出永远是不可信输入

应该像处理浏览器提交的表单一样处理模型输出：

1. 先解析，不直接执行；
2. 验证结构与类型；
3. 检查资源和权限白名单；
4. 限制执行时间与结果大小；
5. 使用最小权限连接；
6. 记录安全、脱敏的审计信息。

模型可能因为理解错误、上下文污染或恶意输入偏离预期。即使当前模型在测试中从未生成危险语句，也不能把概率表现当成安全保证。

SQL 工作流的五道关

SQL 图可以按下面顺序执行：

inspect_schema -> generate_sql -> validate_sql -> execute -> explain

第一关：只提供允许的 Schema

模型不应该看到整个数据库。inspect_schema 只读取：

sales
products
customers

运行记录、工单和 SQLite 内部表不会进入 Prompt。减少上下文本身就是第一道权限边界。

第二关：解析成单条 AST

模型返回的是文本。应用先用 SQLGlot 解析为抽象语法树（AST）：

statements = sqlglot.parse(sql.strip(), read="sqlite")
if len(statements) != 1:
    raise UnsafeSqlError("只允许执行一条 SQL")

用字符串搜索 DELETE 并不可靠，因为大小写、注释、子查询和语法变体都可能绕过简单判断。结构化解析器才能判断语句真正表达了什么。

第三关：只允许查询和白名单表

顶层必须是 SELECT 或 WITH ... SELECT 对应的查询结构，且 AST 中不能包含写入或管理节点：

forbidden = (
    exp.Insert, exp.Update, exp.Delete,
    exp.Create, exp.Drop, exp.Alter, exp.Command,
)
if any(statement.find(kind) is not None for kind in forbidden):
    raise UnsafeSqlError("查询包含写入或管理操作")

随后遍历真正引用的表，排除合法 CTE 名称，再与分析表白名单比较。这样可以阻止模型查询 runs、tickets 或 sqlite_master，也避免有人用同名 CTE 掩盖内部表引用。

第四关：只读连接与执行上限

即使 AST 校验存在缺陷，数据库连接本身仍应是只读的。例如通过 SQLite 只读连接执行，并设置 progress handler 限制执行步数。

这是“纵深防御”：

• AST 校验负责拒绝已知危险结构；
• 表白名单限制数据范围；
• 只读连接阻止写入；
• 执行步数上限阻止失控查询长期占用资源。

不要让一个安全措施承担全部责任。

第五关：最多返回 200 行

没有 LIMIT 时自动添加：

LIMIT 200

已有上限超过 200 时则压低到 200。非整数 LIMIT 直接拒绝。

限制结果集既保护数据库和网络，也避免把大量数据再次交给模型造成成本和隐私风险。

可观察性与失败轨迹

多步骤系统不能只记录最终异常。事件记录器应为每次运行生成 run_id，并按顺序保存节点事件。

成功路径可能是：

node_started(generate_sql)
node_completed(generate_sql)
node_started(validate_sql)
node_completed(validate_sql)
tool_called(execute_readonly_sql)
run_completed

如果 SQL 校验失败，则会记录：

node_started(validate_sql)
run_failed(validate_sql)

这让你能回答：

• 是模型超时，还是数据库超时？
• 哪个节点失败？
• 工具是否真的执行？
• 最近一次修改是否让某个节点变慢？

日志不是越多越好。它应该足以定位问题，同时遵守数据边界。

隐私和日志边界

运行轨迹很容易变成新的敏感数据仓库。Prompt、工具参数和工具结果里可能包含姓名、电话、订单、银行卡号或访问令牌。

可以采用几层保护：

• 密钥与客户字段按字段名脱敏；
• 客服原始输入统一替换为占位符；
• 政策查询和工单参数不记录原始客户消息；
• 字符串、列表、字典大小和嵌套深度都有上限；
• 失败信息转换为安全错误，不返回底层异常细节。

例如客服轨迹只会看到：

{
  "message": "[CUSTOMER MESSAGE REDACTED]"
}

而不会为了“方便调试”保存完整投诉文本。

需要特别注意：浏览器端折叠或截断不是服务端保护。数据如果已经落库并通过 API 传输，前端不显示也来不及了。脱敏和大小限制必须发生在写入之前。

评测比感觉更可靠

模型应用的输出不是简单的布尔值，但仍然可以建立固定评测。

一套最小评测可以包含六个代表性场景：

1. RAG 命中退款政策；
2. RAG 无资料时拒答；
3. 客服按订单号查单；
4. 破损赔偿升级人工；
5. SQL 趋势结果正确；
6. SQL 写操作被拒绝。

这组用例同时检查“应该成功”和“必须拒绝”。安全系统如果只测试正常路径，通常会高估自己的可靠性。

测试应分别覆盖 SQL 安全规则和端到端评测，并在模型、Prompt、切分参数或工具描述发生变化时统一运行。

生产项目还应收集真实失败样本，经过脱敏和人工标注后加入评测集。模型、Prompt、切分参数或工具描述发生变化时，先跑评测再发布。

动手练习

1. 运行 SQL 安全测试，找出多语句、内部表、写操作和超大 LIMIT 分别对应的测试样本。
2. 在 Python REPL 中调用 validate_read_only_sql("SELECT * FROM sales LIMIT 500")，确认结果被限制为 200 行。
3. 运行一次 SQL 分析，从事件存储中核对 generate_sql、validate_sql 和 execute 的顺序。
4. 为你自己的模型应用写一条“必须拒绝”的固定评测，并说明如果它失败会造成什么业务后果。

什么时候不该用 LangChain

下面这些情况，普通 Python 往往更合适：

• 只有一次固定模型调用；
• 没有检索、工具、状态或分支；
• 团队无法为新增抽象建立测试；
• 框架封装反而遮蔽了关键供应商能力；
• 问题可以用确定性程序完整解决，根本不需要模型。

引入 LangChain 的理由应该是它让复杂连接更清楚、更可替换、更可观察，而不是“AI 项目都应该用”。

上线检查清单

在发布一个 LangChain 应用前，至少逐项确认：

• [ ] 模型输出在进入数据库、文件系统或外部 API 前经过结构校验；
• [ ] 每个 Tool 有最小权限、参数 Schema 和业务白名单；
• [ ] 高风险动作需要人工批准或二次确认；
• [ ] Agent 有最大轮数、超时、token 和费用上限；
• [ ] RAG 有引用、拒答和知识库外问题测试；
• [ ] 失败能定位到具体节点；
• [ ] 轨迹在服务端写入前完成脱敏和截断；
• [ ] 错误响应不泄露底层异常或密钥；
• [ ] 固定评测同时覆盖成功和拒绝路径；
• [ ] 模型、Prompt 或依赖升级前运行回归测试。

下一步学习路线

现在你已经有了一条务实路线：

1. 先用官方 SDK 完成单次调用；
2. 用 Runnable 组合可替换的直线步骤；
3. 用 LangGraph 表达状态、分支和恢复；
4. 用 RAG 连接可信资料；
5. 用 Tool 暴露最小业务能力；
6. 只有在步骤无法预先确定时才考虑自主 Agent；
7. 用护栏、轨迹、脱敏和评测把能力包进工程边界。

继续学习时，不妨优先研究两件事：如何构建贴近真实业务的评测集，以及如何让每个模型决策都有可核验的证据。它们通常比再学习一个新 Agent 名词更能改善系统质量。

上一篇：Tools 与 Agent · 返回系列目录