我最近在带一个内部项目代号叫LCODER核心是给业务团队做一个“问数”智能体——让不懂SQL的运营、产品同学直接用自然语言查数。目前第一版已经跑通了端到端流程这里把项目架构的设计过程、技术选型和落地踩坑记录下来。如果你正打算从零搭一个AI Agent项目或者在公司做Text-to-SQL类的智能体这篇应该能帮你少走一些弯路。先说清楚LCODER问数项目到底解决什么问题。传统的取数流程是业务提需求、数据团队写SQL、排期交付一个简单查询动辄等半天。我们的目标是让业务方在对话框里直接问“上周华东区各品类GMV环比变化”智能体自动完成语义理解、表字段匹配、SQL生成、结果执行和可视化返回把整个链路压缩到几秒钟。听起来挺美好但真正落地的时候会发现这不只是“接个大模型API”那么简单——中间涉及多Agent协作、工具调用、知识库召回、会话记忆等一堆设计决策。这篇文章是系列第一篇聚焦在项目架构层讲清楚我们为什么这么拆、每个模块的职责边界以及实际编码时的关键实现。1. 整体设计思路为什么问数场景必须用多Agent而不是单Prompt刚开始设计的时候我也试过“一步到位”——一个大模型调用把数据库Schema、用户问题、few-shot示例全塞进Prompt里让模型直接输出SQL。简单场景下比如两三张表、几十个字段确实能跑但一旦业务库表复杂起来问题马上暴露。1.1 单Prompt方案在复杂问数场景下的三个致命问题第一个问题是上下文爆炸。我们的核心库有几十张表、上千个字段如果全部塞进PromptToken数量轻松破万。这不仅浪费成本更重要的是模型在面对超长上下文时注意力会分散反而更不容易抓住关键信息。实测下来Schema越长SQL生成的准确率反而越低这是个反直觉但真实存在的现象。第二个问题是职责混乱。让一个模型同时做“理解用户意图”、“定位相关表和字段”、“生成SQL”、“判断结果是否合理”这四件事相当于让一个人既当产品经理又当架构师还当测试员。每件事的Prompt要求不一样混在一起必然互相干扰。比如意图理解需要口语化宽松一点的指令SQL生成又需要严格的语法约束这两者很难在同一个Prompt里同时达到最优。第三个问题是不好排查。单Prompt方案一旦出错——模型生成了错误的SQL、字段名写错了、业务条件漏了——你根本不知道问题出在哪个环节。是意图理解错了还是Schema召回不全还是模型本身生成能力不行没有中间过程的状态输出调试基本靠猜。1.2 多Agent编排的核心思路把复杂链路拆成可观测的节点LCODER最终采用了多Agent编排的方案核心思路是按“认知链路”拆节点每个节点只干一件事节点之间通过结构化状态传递信息。整体流程拆成五个节点意图识别、Schema召回、SQL生成、SQL校验、结果解释。每个节点都是一个独立Agent拥有自己的Prompt、模型参数和工具权限。节点之间不直接对话而是通过一个共享的State对象传递数据。这样做的好处特别明显第一每个节点都很轻Prompt不会互相污染第二任何一个节点出错都能在State里看到完整的输入输出定位问题非常快第三后续要替换某个节点的实现比如换一个更好的模型只需要动局部不影响整体流程。用LangGraph来编排这套流程是基于它的图执行机制。LangGraph天然支持节点之间有条件的跳转——比如SQL校验不通过可以回退到SQL生成节点重新生成形成一个闭环。这种“校验-重试”的循环机制是解决模型生成不稳定问题的关键武器。1.3 问数Agent的典型请求生命周期拿一个真实请求走一遍流程你就能感受到这套架构的价值。假设用户问“统计上个月每个城市的订单量Top10”。State先记录原始问题进入意图识别节点模型判断用户想做的是一张分组统计报表目标表是订单表时间范围是上个月。接着Schema召回节点上场它不去看全库的表结构而是先通过向量检索找到跟“订单量”、“城市”最相关的表和字段把完整的表结构、字段注释、枚举值拼成一个精炼的Schema子集。SQL生成节点拿到这个子集配合几个动态挑选出来的few-shot示例生成初步SQL。SQL校验节点会做三件事语法校验能不能跑、语义校验字段名和表名是否存在、安全校验是不是只有查询操作、有没有超时风险。如果校验不通过带上错误信息回退到生成节点重试如果通过了就在只读账号下执行。最后结果解释节点把查询结果翻译成自然语言附上表格或图表返回给用户。2. 技术栈选型LangGraph、模型接入与向量检索方案架构定了之后选型就是下一个大坑。工具链选得好开发和维护事半功倍选得不好后面全是泪。2.1 编排框架为什么选LangGraph而不是LangChain或自研框架我见过不少团队直接用LangChain搭Agent快速原型确实方便但问题出在可控性和状态管理上。LangChain的AgentExecutor本质上是一个循环模型决定调哪个工具、再调、再看结果整个过程是黑盒的你想插入一个中间校验步骤非常别扭。LangGraph给出的模型是“图”每个节点是一个Function每条边是节点间的跳转逻辑。这一切都是显式的代码不是隐式的Prompt。你可以完全控制流程走向——什么条件下重试、什么条件下终止、什么条件下调用人类审批。对于问数这种对准确性要求很高的场景这种显式控制几乎是必须的。而且LangGraph对状态管理做了内置支持。你可以定义一个TypedDict作为所有节点共享的State每个节点读自己需要的字段、写自己产出的字段LangGraph负责把这些字段状态在不同节点间传递。这种设计让每个节点变得很纯粹——输入是State输出是State的局部更新完全符合单一职责原则。2.2 模型接入与多模型路由策略模型选型上我们用了双模型策略。意图识别和Schema召回节点需要的是快速、便宜、上下文窗口适中我们用了一个中等规模的模型Qwen系列的中杯规格单次调用成本低、延迟在几百毫秒以内。SQL生成节点是重头戏需要最强的SQL理解和生成能力这里用大杯规格的模型且Temperature调到接近0保证输出尽可能确定。实测下来SQL生成节点的模型能力直接决定整个项目的天花板。如果你用的小模型总是生成字段名错误、函数用错别指望后续节点能兜住——校验节点只能发现问题并打回去重试但重试多次都失败的话用户体验会非常差。我们花了大量时间在模型对比和Prompt调优上这是值得的。模型接入是通过OpenAI兼容接口做的好处是未来换模型厂商零成本。业务侧不直接依赖具体模型品牌只定义一个ModelFactory根据配置创建不同的LLM实例。这个抽象层看起来微不足道但在模型日新月异的当下能让你跟进新模型非常从容。2.3 Schema召回向量检索与倒排索引结合Schema召回是整个链路里最容易被低估的环节。很多人以为直接全量Schema丢给模型就行但前面说了库一复杂就崩。我们的做法是先做一层相关性召回把“用户问题”映射到“可能相关的表和字段”。具体实现是用Embedding模型把表注释、字段注释、枚举值等信息向量化存入向量数据库。用户发起查询时把问题也做向量化做TopK相似度检索。但仅靠向量检索有召回不全的问题——比如用户说“用户数”但表里叫“UV”语义相近向量能召回可有些术语是业务独有的缩写向量模型没见过。所以我们在向量检索之外还叠加了一层基于倒排索引的关键词匹配双路召回后合并取并集。这个方案跑下来基本满足需求但在Schema召回里其实还可以做得更细——对列字段的关键性加权。例如主键、外键、时间字段、聚合指标字段的召回权重应该不一样。我们在后续迭代里会把这个提上日程。2.4 工具层MCP协议与内置工具的取舍工具层是用来扩展Agent能力的。LCODER的问数Agent至少需要三类工具数据库查询执行器、可视化生成工具、以及后续要接的钉钉/飞书消息通知工具。这里我们做了个关键选择工具定义走MCP协议标准的示例风格同时自研轻量工具注册框架。MCPModel Context Protocol的目标是让工具跟Agent之间有一个标准化的交互方式工具描述、输入输出Schema、执行入口都统一格式。好处是生态互通——未来如果社区有更好的工具实现可以直接替换。但现阶段MCP的生态还不够丰富所以我们选择利用MCP的思想自己实现了工具注册和调用的框架兼容后续平滑迁移到MCP完整协议。工具层设计的关键是“描述文档优先”。Agent能不能正确调用工具完全取决于工具描述写得好不好。很多项目在这块偷懒工具描述写得很粗糙比如“查询数据库”四个字模型根本不知道这个工具能不能执行SQL、有什么限制、返回什么格式。我们每一份工具描述里都包含功能说明、参数列表及类型约束、典型使用范例、错误返回格式。在问数Agent里光数据库执行器就写了二百多字的描述说明这在纯技术工具里已经算是相当细的了。3. 项目架构分层接入层、编排层、知识层与存储层回到项目架构的整体图景。LCODER问数项目按层次划分从下到上分别是存储层、知识层、编排层、接入层。每一层各司其职上层依赖下层的接口但不跨层调用。3.1 接入层对外暴露什么样的交互形态接入层解决的是“用户从哪里发起问题”。我们同时支持了三端Web聊天界面基于Streamlit快速搭的Demo、企业微信机器人业务方使用频率最高、以及OpenAPI接口给其他系统集成用。架构上接入层的设计要点是协议统一。无论从哪个端进来最终都转换成一个统一的AgentRequest对象包含对话ID、用户ID、消息内容、上下文历史等字段再交给编排层处理。响应用同样统一成AgentResponse包含最终回复、中间过程日志、引用的数据表、生成的SQL等元信息。这个协议统一层花了我们不少功夫。因为不同端的信息结构差异很大——企业微信消息有Event结构、Web聊天有Session概念、API调用有鉴权Token——如果每端自己写一套解析逻辑后面维护三个端就是维护三份代码。统一之后新增一个渠道只需要写一个几十行的适配器。3.2 编排层LangGraph图定义、节点注册与状态流转编排层是整个架构的心脏。我们在代码里用LangGraph定义了一个StateGraph核心节点就是前面提到的五个。每个节点都实现一个统一的接口接收State处理后返回State的部分更新。State对象的设计也是迭代了几版才稳定。第一版把所有字段都平铺在State里流转起来很乱后来改成嵌套结构分成了query_state用户问题和意图信息、schema_state召回的Schema信息、sql_state生成的SQL和校验状态、result_state执行结果和解释文案四个子模块。这样每个节点只关注自己负责的子模块不容易误改别人的数据。图定义里还加入了条件边。比如SQL校验节点如果返回“不通过且重试次数小于3次”就回退到SQL生成节点重新生成如果重试次数超限就跳到一个兜底节点回复用户“暂时无法处理该查询请尝试换个问法或联系数据团队”。这个兜底逻辑非常重要否则用户在Agent出错时会陷入无响应的迷茫。3.3 知识层元数据管理、Schema向量索引与Few-shot库知识层是问数Agent跟通用ChatBot最大的区别。通用ChatBot的知识主要是文档和百科问数Agent的知识则是数据库的元数据——表结构、字段含义、枚举值、表间关系、常用查询模板。元数据管理模块负责定时从数据库同步表结构信息同步后做三件事一是更新Schema向量索引二是更新倒排索引三是生成Schema摘要存到缓存里用于快速检索预览。同步任务每天凌晨跑一次因为表结构变更不会特别频繁日更足够了。Few-shot库是另一个容易忽略但特别重要的组件。同样是“按城市分组统计订单量”这个需求模型看着一句示例SQL做出来的结果和没有示例硬编出来的结果准确率差距巨大。我们从历史正确查询里沉淀了一批高质量样例按查询类型趋势分析、占比分析、TopN排名、对比分析等打标存储SQL生成节点按需取用。3.4 存储层会话状态、向量库、日志与审计存储层包含四类存储。第一是Redis用来存短期会话状态和上下文窗口Key格式是session:{session_id}自动过期时间8小时保证对话有上下文但不过度堆积。第二是向量数据库存Schema向量和Few-shot向量这里我们选型考虑了运维成本最终用了轻量方案如果你所在公司有条件上专业向量库建议用更大的分布式方案方便后续扩展。第三是查询日志库记录每一次用户查询、生成的SQL、执行耗时、结果状态这是审计和排查的底牌。第四是结果缓存短期高频查询比如“昨日GMV”如果结果已经算过且源表没更新直接命中缓存返回节约模型调用和执行成本。存储层的设计原则是能缓存就缓存每省一次模型调用都是实打实的成本。4. 核心模块拆解与代码落地这一节直接上代码讲清楚核心节点怎么实现目录怎么组织方便你直接参考落地。4.1 工程目录结构与模块划分lcoder-agent/ ├── agent/ # Agent核心编排逻辑 │ ├── graph.py # LangGraph图定义与节点装配 │ ├── state.py # State数据结构定义 │ ├── nodes/ # 各节点实现 │ │ ├── intent.py # 意图识别节点 │ │ ├── schema_recall.py # Schema召回节点 │ │ ├── sql_gen.py # SQL生成节点 │ │ ├── sql_validate.py # SQL校验节点 │ │ └── result_explain.py # 结果解释节点 │ └── tools/ # 工具注册与实现 │ ├── registry.py # 工具注册器 │ ├── db_executor.py # 数据库执行工具 │ └── viz_generator.py # 可视化工具 ├── knowledge/ # 知识层 │ ├── schema_sync.py # 元数据同步任务 │ ├── vector_index.py # Schema向量化与检索 │ └── few_shot.py # Few-shot样例管理 ├── storage/ # 存储层 │ ├── redis_store.py # 会话状态存储 │ ├── log_store.py # 查询日志存储 │ └── cache.py # 结果缓存 ├── api/ # 接入层 │ ├── server.py # FastAPI服务入口 │ └── adapters/ # 渠道适配器Web/企微/API └── config/ # 配置中心 ├── settings.py # 全局配置 └── prompts/ # 各节点Prompt模板目录分层的核心逻辑是依赖方向——knowledge层不知道api层存在agent层只依赖knowledge和storage的接口api层再调agent层。这样每一层都能独立测试迭代时改动一片不影响其他。4.2 核心代码State定义与图装配State定义用的pydantic BaseModel类型标注要尽量精确因为你后面写节点函数时要依赖类型的自动校验。from datetime import datetime from typing import Optional from pydantic import BaseModel, Field class QueryState(BaseModel): original_query: str user_id: str session_id: str intent: Optional[str] None query_type: Optional[str] None # 趋势/占比/TopN/对比等 class SchemaState(BaseModel): recalled_tables: list Field(default_factorylist) recalled_columns: list Field(default_factorylist) schema_prompt_block: Optional[str] None class SQLState(BaseModel): generated_sql: Optional[str] None retry_count: int 0 validation_result: Optional[dict] None class ResultState(BaseModel): query_result: Optional[list] None error_message: Optional[str] None explain_text: Optional[str] None class AgentState(BaseModel): query: QueryState schema: SchemaState Field(default_factorySchemaState) sql: SQLState Field(default_factorySQLState) result: ResultState Field(default_factoryResultState) meta: dict Field(default_factorydict)接着把五个节点函数装配成图。每个节点函数签名都是(state: AgentState) - dict返回dict会被自动合并回state实现局部更新。from langgraph.graph import StateGraph, END def build_agent_graph(): workflow StateGraph(AgentState) workflow.add_node(intent, intent_node) workflow.add_node(schema_recall, schema_recall_node) workflow.add_node(sql_gen, sql_gen_node) workflow.add_node(sql_validate, sql_validate_node) workflow.add_node(result_explain, result_explain_node) workflow.set_entry_point(intent) workflow.add_edge(intent, schema_recall) workflow.add_edge(schema_recall, sql_gen) workflow.add_edge(sql_gen, sql_validate) # 条件边校验不通过且重试次数未满则回流SQL生成 workflow.add_conditional_edges( sql_validate, route_after_validate, { regenerate: sql_gen, explain: result_explain, giveup: result_explain } ) workflow.add_edge(result_explain, END) return workflow.compile() def route_after_validate(state: AgentState) - str: if state.sql.validation_result and state.sql.validation_result.get(passed): return explain if state.sql.retry_count 3: state.result.error_message SQL校验多次未通过请尝试换个问法 return giveup return regenerate这里的条件边设计是整个图跟普通链式调用的本质区别。你在代码里明确写出“校验不过就打回重新生成”这是LangChain那种AgentExecutor做不到的。4.3 SQL生成节点的Prompt组织技巧SQL生成节点是整个项目最需要抠细节的地方。Prompt组织按照四段式来搭角色设定、Schema信息、Few-shot示例、用户问题与约束。SQL_GEN_PROMPT_TEMPLATE 你是一名资深数据分析师根据提供的数据表信息为用户的查询需求编写一条SQL。 【数据表信息】 {schema_block} 【参考示例】 {few_shot_block} 【用户需求】 {user_query} 【约束条件】 1. 只允许SELECT查询 2. 必须使用WHERE条件避免全表扫描 3. 时间字段统一使用字段名 4. 如果用户指定了“最近X天”用DATE_ADD(CURRENT_DATE, INTERVAL -X DAY) 5. 只输出SQL不要输出任何解释 几个值得注意的细节。第一约束条件必须写清楚“只输出SQL”否则模型会附带一堆读起来费力的解释增加解析成本和出错概率。第二few-shot的选择不是随机的而是根据识别出的query_type去Few-shot库里挑选同类型的样例。第三Schema Block一定要裁剪到只包含召回模块检索出来的表和字段不要贪多。实践里还有一个非常管用的技巧在Schema信息里给关键字段添加“数据特征注释”。比如“字段 status: 取值范围[0待支付,1已支付,2已取消]”。模型看到枚举值生成的WHERE条件会准确很多还能避免一些字段理解歧义。4.4 SQL校验节点的三层校验实现SQL校验节点绝对不能省。模型生成的SQL直接丢给数据库执行轻则报错浪费一次调用重则可能跑出一个超大查询把数据库拖垮。我们实现了三层校验。第一层是静态语法校验用sqlparse解析SQL检查基本的语法结构再检查关键字白名单。第二层是元数据校验从SQL里提取所有表名和字段名到元数据服务里比对确认都存在且有权访问。第三层是安全与成本预估检查是否有LIMIT子句、WHERE条件是否强制存在、预估扫描行数是否在阈值内。import sqlparse import re def validate_sql(sql: str, meta_service) - dict: # 第一层语法与安全 if not sqlparse.parse(sql): return {passed: False, reason: SQL语法无法解析} normalized sql.upper() forbidden [INSERT, UPDATE, DELETE, DROP, ALTER, TRUNCATE] for kw in forbidden: if kw in normalized: return {passed: False, reason: f包含禁止关键字: {kw}} # 第二层元数据匹配 tables extract_tables(sql) for t in tables: if not meta_service.table_exists(t): return {passed: False, reason: f表不存在: {t}} # 第三层执行保护 if WHERE not in normalized: return {passed: False, reason: 缺少WHERE条件可能导致全表扫描} if LIMIT not in normalized: if normalized.count(SELECT) 1: # 自动追加LIMIT这里根据实际需求可让模型重做或直接改写 sql sql.rstrip().rstrip(;) LIMIT 100 return {passed: True, sql: sql}拦截掉危险SQL的功劳主要在这一层。出现过一次模型把“近30天退款订单数”理解成对退款表做DROP的极端情况——校验层直接拦下来回去重生成避免了风险。4.5 会话记忆如何在多轮对话中维持查询上下文问数场景天然是多轮对话用户第一轮问“上个月各品类GMV”第二轮接着说“只看华东区”你要能理解“只”字是在第一轮的结果上做条件过滤。这要求Agent有会话记忆能力。我们的实现是用Redis存最近N轮对话摘要和最近一轮的SQL执行结果。具体做法是每轮结束后把用户问题和最终生成的SQL压缩成一个结构化记录存进Redis。下一轮追问进来取出历史记录拼接到当前Prompt的上下文位置。注意拼接的是“历史问题对应SQL”而不是把所有历史原文全塞进去这样既保留了执行上下文又不至于Token爆炸。会话记忆还有一个容易被忽略的作用——结果对比。用户问“上周”之后继续问“那这周呢”你需要在结果解释节点里拿到上周的结果跟本周做对比分析。所以State的ResultState里始终保留上轮执行结果的摘要这个设计在真实使用中好评度很高。5. 架构演进与后续规划这套架构上线跑了两周整体稳定性比预期好但暴露出的问题也值得记录。第一个问题是Schema召回在复杂查询时不够精准比如用户问“高价值用户”系统只召回“用户”相关表但真正有价值的信息分散在“订单”和“用户等级”两张表里需要多表关联。这个问题单靠向量检索很难解决我们的方案是提前在知识层维护“业务主题-多表关联关系”的映射做一次业务层面的预聚合建模。第二个问题是长任务的处理。当前架构是同步请求-响应模式用户等几秒钟能拿到结果。但有些复杂查询比如跨月、跨表、大聚合可能要跑几十秒同步等待的体验就很差。后续规划是把任务队列和异步通知机制引入复杂查询先提交任务完成后通过企微机器人推送结果。第三个方向是MCP协议的全面适配。我们已经按MCP的思路设计了工具层后续等生态成熟可以直接切换。接入更多数据源ClickHouse、Elasticsearch时工具层只要能注册对应执行器就行架构上不用做大改动。再往后我们希望加入Agent的自我学习能力——把每轮“校验失败-重试成功”的案例自动沉淀进Few-shot库让系统越用越聪明。这个在评估安全性比如防止用户通过SQL注入给系统投毒之后才会动工。问数Agent的架构设计本质上是在“模型能力边界”和“业务对准确性的要求”之间找平衡。模型再强也会偶尔出错但通过合理的架构设计——拆节点、加校验、做召回、带记忆——能把出错率压到可以接受的范围。这也是我想通过这系列文章传达的核心思路架构不是炫技而是拿来对抗模型不确定性的。
从零搭建Text-to-SQL问数智能体：LCODER项目架构设计与实践
我最近在带一个内部项目代号叫LCODER核心是给业务团队做一个“问数”智能体——让不懂SQL的运营、产品同学直接用自然语言查数。目前第一版已经跑通了端到端流程这里把项目架构的设计过程、技术选型和落地踩坑记录下来。如果你正打算从零搭一个AI Agent项目或者在公司做Text-to-SQL类的智能体这篇应该能帮你少走一些弯路。先说清楚LCODER问数项目到底解决什么问题。传统的取数流程是业务提需求、数据团队写SQL、排期交付一个简单查询动辄等半天。我们的目标是让业务方在对话框里直接问“上周华东区各品类GMV环比变化”智能体自动完成语义理解、表字段匹配、SQL生成、结果执行和可视化返回把整个链路压缩到几秒钟。听起来挺美好但真正落地的时候会发现这不只是“接个大模型API”那么简单——中间涉及多Agent协作、工具调用、知识库召回、会话记忆等一堆设计决策。这篇文章是系列第一篇聚焦在项目架构层讲清楚我们为什么这么拆、每个模块的职责边界以及实际编码时的关键实现。1. 整体设计思路为什么问数场景必须用多Agent而不是单Prompt刚开始设计的时候我也试过“一步到位”——一个大模型调用把数据库Schema、用户问题、few-shot示例全塞进Prompt里让模型直接输出SQL。简单场景下比如两三张表、几十个字段确实能跑但一旦业务库表复杂起来问题马上暴露。1.1 单Prompt方案在复杂问数场景下的三个致命问题第一个问题是上下文爆炸。我们的核心库有几十张表、上千个字段如果全部塞进PromptToken数量轻松破万。这不仅浪费成本更重要的是模型在面对超长上下文时注意力会分散反而更不容易抓住关键信息。实测下来Schema越长SQL生成的准确率反而越低这是个反直觉但真实存在的现象。第二个问题是职责混乱。让一个模型同时做“理解用户意图”、“定位相关表和字段”、“生成SQL”、“判断结果是否合理”这四件事相当于让一个人既当产品经理又当架构师还当测试员。每件事的Prompt要求不一样混在一起必然互相干扰。比如意图理解需要口语化宽松一点的指令SQL生成又需要严格的语法约束这两者很难在同一个Prompt里同时达到最优。第三个问题是不好排查。单Prompt方案一旦出错——模型生成了错误的SQL、字段名写错了、业务条件漏了——你根本不知道问题出在哪个环节。是意图理解错了还是Schema召回不全还是模型本身生成能力不行没有中间过程的状态输出调试基本靠猜。1.2 多Agent编排的核心思路把复杂链路拆成可观测的节点LCODER最终采用了多Agent编排的方案核心思路是按“认知链路”拆节点每个节点只干一件事节点之间通过结构化状态传递信息。整体流程拆成五个节点意图识别、Schema召回、SQL生成、SQL校验、结果解释。每个节点都是一个独立Agent拥有自己的Prompt、模型参数和工具权限。节点之间不直接对话而是通过一个共享的State对象传递数据。这样做的好处特别明显第一每个节点都很轻Prompt不会互相污染第二任何一个节点出错都能在State里看到完整的输入输出定位问题非常快第三后续要替换某个节点的实现比如换一个更好的模型只需要动局部不影响整体流程。用LangGraph来编排这套流程是基于它的图执行机制。LangGraph天然支持节点之间有条件的跳转——比如SQL校验不通过可以回退到SQL生成节点重新生成形成一个闭环。这种“校验-重试”的循环机制是解决模型生成不稳定问题的关键武器。1.3 问数Agent的典型请求生命周期拿一个真实请求走一遍流程你就能感受到这套架构的价值。假设用户问“统计上个月每个城市的订单量Top10”。State先记录原始问题进入意图识别节点模型判断用户想做的是一张分组统计报表目标表是订单表时间范围是上个月。接着Schema召回节点上场它不去看全库的表结构而是先通过向量检索找到跟“订单量”、“城市”最相关的表和字段把完整的表结构、字段注释、枚举值拼成一个精炼的Schema子集。SQL生成节点拿到这个子集配合几个动态挑选出来的few-shot示例生成初步SQL。SQL校验节点会做三件事语法校验能不能跑、语义校验字段名和表名是否存在、安全校验是不是只有查询操作、有没有超时风险。如果校验不通过带上错误信息回退到生成节点重试如果通过了就在只读账号下执行。最后结果解释节点把查询结果翻译成自然语言附上表格或图表返回给用户。2. 技术栈选型LangGraph、模型接入与向量检索方案架构定了之后选型就是下一个大坑。工具链选得好开发和维护事半功倍选得不好后面全是泪。2.1 编排框架为什么选LangGraph而不是LangChain或自研框架我见过不少团队直接用LangChain搭Agent快速原型确实方便但问题出在可控性和状态管理上。LangChain的AgentExecutor本质上是一个循环模型决定调哪个工具、再调、再看结果整个过程是黑盒的你想插入一个中间校验步骤非常别扭。LangGraph给出的模型是“图”每个节点是一个Function每条边是节点间的跳转逻辑。这一切都是显式的代码不是隐式的Prompt。你可以完全控制流程走向——什么条件下重试、什么条件下终止、什么条件下调用人类审批。对于问数这种对准确性要求很高的场景这种显式控制几乎是必须的。而且LangGraph对状态管理做了内置支持。你可以定义一个TypedDict作为所有节点共享的State每个节点读自己需要的字段、写自己产出的字段LangGraph负责把这些字段状态在不同节点间传递。这种设计让每个节点变得很纯粹——输入是State输出是State的局部更新完全符合单一职责原则。2.2 模型接入与多模型路由策略模型选型上我们用了双模型策略。意图识别和Schema召回节点需要的是快速、便宜、上下文窗口适中我们用了一个中等规模的模型Qwen系列的中杯规格单次调用成本低、延迟在几百毫秒以内。SQL生成节点是重头戏需要最强的SQL理解和生成能力这里用大杯规格的模型且Temperature调到接近0保证输出尽可能确定。实测下来SQL生成节点的模型能力直接决定整个项目的天花板。如果你用的小模型总是生成字段名错误、函数用错别指望后续节点能兜住——校验节点只能发现问题并打回去重试但重试多次都失败的话用户体验会非常差。我们花了大量时间在模型对比和Prompt调优上这是值得的。模型接入是通过OpenAI兼容接口做的好处是未来换模型厂商零成本。业务侧不直接依赖具体模型品牌只定义一个ModelFactory根据配置创建不同的LLM实例。这个抽象层看起来微不足道但在模型日新月异的当下能让你跟进新模型非常从容。2.3 Schema召回向量检索与倒排索引结合Schema召回是整个链路里最容易被低估的环节。很多人以为直接全量Schema丢给模型就行但前面说了库一复杂就崩。我们的做法是先做一层相关性召回把“用户问题”映射到“可能相关的表和字段”。具体实现是用Embedding模型把表注释、字段注释、枚举值等信息向量化存入向量数据库。用户发起查询时把问题也做向量化做TopK相似度检索。但仅靠向量检索有召回不全的问题——比如用户说“用户数”但表里叫“UV”语义相近向量能召回可有些术语是业务独有的缩写向量模型没见过。所以我们在向量检索之外还叠加了一层基于倒排索引的关键词匹配双路召回后合并取并集。这个方案跑下来基本满足需求但在Schema召回里其实还可以做得更细——对列字段的关键性加权。例如主键、外键、时间字段、聚合指标字段的召回权重应该不一样。我们在后续迭代里会把这个提上日程。2.4 工具层MCP协议与内置工具的取舍工具层是用来扩展Agent能力的。LCODER的问数Agent至少需要三类工具数据库查询执行器、可视化生成工具、以及后续要接的钉钉/飞书消息通知工具。这里我们做了个关键选择工具定义走MCP协议标准的示例风格同时自研轻量工具注册框架。MCPModel Context Protocol的目标是让工具跟Agent之间有一个标准化的交互方式工具描述、输入输出Schema、执行入口都统一格式。好处是生态互通——未来如果社区有更好的工具实现可以直接替换。但现阶段MCP的生态还不够丰富所以我们选择利用MCP的思想自己实现了工具注册和调用的框架兼容后续平滑迁移到MCP完整协议。工具层设计的关键是“描述文档优先”。Agent能不能正确调用工具完全取决于工具描述写得好不好。很多项目在这块偷懒工具描述写得很粗糙比如“查询数据库”四个字模型根本不知道这个工具能不能执行SQL、有什么限制、返回什么格式。我们每一份工具描述里都包含功能说明、参数列表及类型约束、典型使用范例、错误返回格式。在问数Agent里光数据库执行器就写了二百多字的描述说明这在纯技术工具里已经算是相当细的了。3. 项目架构分层接入层、编排层、知识层与存储层回到项目架构的整体图景。LCODER问数项目按层次划分从下到上分别是存储层、知识层、编排层、接入层。每一层各司其职上层依赖下层的接口但不跨层调用。3.1 接入层对外暴露什么样的交互形态接入层解决的是“用户从哪里发起问题”。我们同时支持了三端Web聊天界面基于Streamlit快速搭的Demo、企业微信机器人业务方使用频率最高、以及OpenAPI接口给其他系统集成用。架构上接入层的设计要点是协议统一。无论从哪个端进来最终都转换成一个统一的AgentRequest对象包含对话ID、用户ID、消息内容、上下文历史等字段再交给编排层处理。响应用同样统一成AgentResponse包含最终回复、中间过程日志、引用的数据表、生成的SQL等元信息。这个协议统一层花了我们不少功夫。因为不同端的信息结构差异很大——企业微信消息有Event结构、Web聊天有Session概念、API调用有鉴权Token——如果每端自己写一套解析逻辑后面维护三个端就是维护三份代码。统一之后新增一个渠道只需要写一个几十行的适配器。3.2 编排层LangGraph图定义、节点注册与状态流转编排层是整个架构的心脏。我们在代码里用LangGraph定义了一个StateGraph核心节点就是前面提到的五个。每个节点都实现一个统一的接口接收State处理后返回State的部分更新。State对象的设计也是迭代了几版才稳定。第一版把所有字段都平铺在State里流转起来很乱后来改成嵌套结构分成了query_state用户问题和意图信息、schema_state召回的Schema信息、sql_state生成的SQL和校验状态、result_state执行结果和解释文案四个子模块。这样每个节点只关注自己负责的子模块不容易误改别人的数据。图定义里还加入了条件边。比如SQL校验节点如果返回“不通过且重试次数小于3次”就回退到SQL生成节点重新生成如果重试次数超限就跳到一个兜底节点回复用户“暂时无法处理该查询请尝试换个问法或联系数据团队”。这个兜底逻辑非常重要否则用户在Agent出错时会陷入无响应的迷茫。3.3 知识层元数据管理、Schema向量索引与Few-shot库知识层是问数Agent跟通用ChatBot最大的区别。通用ChatBot的知识主要是文档和百科问数Agent的知识则是数据库的元数据——表结构、字段含义、枚举值、表间关系、常用查询模板。元数据管理模块负责定时从数据库同步表结构信息同步后做三件事一是更新Schema向量索引二是更新倒排索引三是生成Schema摘要存到缓存里用于快速检索预览。同步任务每天凌晨跑一次因为表结构变更不会特别频繁日更足够了。Few-shot库是另一个容易忽略但特别重要的组件。同样是“按城市分组统计订单量”这个需求模型看着一句示例SQL做出来的结果和没有示例硬编出来的结果准确率差距巨大。我们从历史正确查询里沉淀了一批高质量样例按查询类型趋势分析、占比分析、TopN排名、对比分析等打标存储SQL生成节点按需取用。3.4 存储层会话状态、向量库、日志与审计存储层包含四类存储。第一是Redis用来存短期会话状态和上下文窗口Key格式是session:{session_id}自动过期时间8小时保证对话有上下文但不过度堆积。第二是向量数据库存Schema向量和Few-shot向量这里我们选型考虑了运维成本最终用了轻量方案如果你所在公司有条件上专业向量库建议用更大的分布式方案方便后续扩展。第三是查询日志库记录每一次用户查询、生成的SQL、执行耗时、结果状态这是审计和排查的底牌。第四是结果缓存短期高频查询比如“昨日GMV”如果结果已经算过且源表没更新直接命中缓存返回节约模型调用和执行成本。存储层的设计原则是能缓存就缓存每省一次模型调用都是实打实的成本。4. 核心模块拆解与代码落地这一节直接上代码讲清楚核心节点怎么实现目录怎么组织方便你直接参考落地。4.1 工程目录结构与模块划分lcoder-agent/ ├── agent/ # Agent核心编排逻辑 │ ├── graph.py # LangGraph图定义与节点装配 │ ├── state.py # State数据结构定义 │ ├── nodes/ # 各节点实现 │ │ ├── intent.py # 意图识别节点 │ │ ├── schema_recall.py # Schema召回节点 │ │ ├── sql_gen.py # SQL生成节点 │ │ ├── sql_validate.py # SQL校验节点 │ │ └── result_explain.py # 结果解释节点 │ └── tools/ # 工具注册与实现 │ ├── registry.py # 工具注册器 │ ├── db_executor.py # 数据库执行工具 │ └── viz_generator.py # 可视化工具 ├── knowledge/ # 知识层 │ ├── schema_sync.py # 元数据同步任务 │ ├── vector_index.py # Schema向量化与检索 │ └── few_shot.py # Few-shot样例管理 ├── storage/ # 存储层 │ ├── redis_store.py # 会话状态存储 │ ├── log_store.py # 查询日志存储 │ └── cache.py # 结果缓存 ├── api/ # 接入层 │ ├── server.py # FastAPI服务入口 │ └── adapters/ # 渠道适配器Web/企微/API └── config/ # 配置中心 ├── settings.py # 全局配置 └── prompts/ # 各节点Prompt模板目录分层的核心逻辑是依赖方向——knowledge层不知道api层存在agent层只依赖knowledge和storage的接口api层再调agent层。这样每一层都能独立测试迭代时改动一片不影响其他。4.2 核心代码State定义与图装配State定义用的pydantic BaseModel类型标注要尽量精确因为你后面写节点函数时要依赖类型的自动校验。from datetime import datetime from typing import Optional from pydantic import BaseModel, Field class QueryState(BaseModel): original_query: str user_id: str session_id: str intent: Optional[str] None query_type: Optional[str] None # 趋势/占比/TopN/对比等 class SchemaState(BaseModel): recalled_tables: list Field(default_factorylist) recalled_columns: list Field(default_factorylist) schema_prompt_block: Optional[str] None class SQLState(BaseModel): generated_sql: Optional[str] None retry_count: int 0 validation_result: Optional[dict] None class ResultState(BaseModel): query_result: Optional[list] None error_message: Optional[str] None explain_text: Optional[str] None class AgentState(BaseModel): query: QueryState schema: SchemaState Field(default_factorySchemaState) sql: SQLState Field(default_factorySQLState) result: ResultState Field(default_factoryResultState) meta: dict Field(default_factorydict)接着把五个节点函数装配成图。每个节点函数签名都是(state: AgentState) - dict返回dict会被自动合并回state实现局部更新。from langgraph.graph import StateGraph, END def build_agent_graph(): workflow StateGraph(AgentState) workflow.add_node(intent, intent_node) workflow.add_node(schema_recall, schema_recall_node) workflow.add_node(sql_gen, sql_gen_node) workflow.add_node(sql_validate, sql_validate_node) workflow.add_node(result_explain, result_explain_node) workflow.set_entry_point(intent) workflow.add_edge(intent, schema_recall) workflow.add_edge(schema_recall, sql_gen) workflow.add_edge(sql_gen, sql_validate) # 条件边校验不通过且重试次数未满则回流SQL生成 workflow.add_conditional_edges( sql_validate, route_after_validate, { regenerate: sql_gen, explain: result_explain, giveup: result_explain } ) workflow.add_edge(result_explain, END) return workflow.compile() def route_after_validate(state: AgentState) - str: if state.sql.validation_result and state.sql.validation_result.get(passed): return explain if state.sql.retry_count 3: state.result.error_message SQL校验多次未通过请尝试换个问法 return giveup return regenerate这里的条件边设计是整个图跟普通链式调用的本质区别。你在代码里明确写出“校验不过就打回重新生成”这是LangChain那种AgentExecutor做不到的。4.3 SQL生成节点的Prompt组织技巧SQL生成节点是整个项目最需要抠细节的地方。Prompt组织按照四段式来搭角色设定、Schema信息、Few-shot示例、用户问题与约束。SQL_GEN_PROMPT_TEMPLATE 你是一名资深数据分析师根据提供的数据表信息为用户的查询需求编写一条SQL。 【数据表信息】 {schema_block} 【参考示例】 {few_shot_block} 【用户需求】 {user_query} 【约束条件】 1. 只允许SELECT查询 2. 必须使用WHERE条件避免全表扫描 3. 时间字段统一使用字段名 4. 如果用户指定了“最近X天”用DATE_ADD(CURRENT_DATE, INTERVAL -X DAY) 5. 只输出SQL不要输出任何解释 几个值得注意的细节。第一约束条件必须写清楚“只输出SQL”否则模型会附带一堆读起来费力的解释增加解析成本和出错概率。第二few-shot的选择不是随机的而是根据识别出的query_type去Few-shot库里挑选同类型的样例。第三Schema Block一定要裁剪到只包含召回模块检索出来的表和字段不要贪多。实践里还有一个非常管用的技巧在Schema信息里给关键字段添加“数据特征注释”。比如“字段 status: 取值范围[0待支付,1已支付,2已取消]”。模型看到枚举值生成的WHERE条件会准确很多还能避免一些字段理解歧义。4.4 SQL校验节点的三层校验实现SQL校验节点绝对不能省。模型生成的SQL直接丢给数据库执行轻则报错浪费一次调用重则可能跑出一个超大查询把数据库拖垮。我们实现了三层校验。第一层是静态语法校验用sqlparse解析SQL检查基本的语法结构再检查关键字白名单。第二层是元数据校验从SQL里提取所有表名和字段名到元数据服务里比对确认都存在且有权访问。第三层是安全与成本预估检查是否有LIMIT子句、WHERE条件是否强制存在、预估扫描行数是否在阈值内。import sqlparse import re def validate_sql(sql: str, meta_service) - dict: # 第一层语法与安全 if not sqlparse.parse(sql): return {passed: False, reason: SQL语法无法解析} normalized sql.upper() forbidden [INSERT, UPDATE, DELETE, DROP, ALTER, TRUNCATE] for kw in forbidden: if kw in normalized: return {passed: False, reason: f包含禁止关键字: {kw}} # 第二层元数据匹配 tables extract_tables(sql) for t in tables: if not meta_service.table_exists(t): return {passed: False, reason: f表不存在: {t}} # 第三层执行保护 if WHERE not in normalized: return {passed: False, reason: 缺少WHERE条件可能导致全表扫描} if LIMIT not in normalized: if normalized.count(SELECT) 1: # 自动追加LIMIT这里根据实际需求可让模型重做或直接改写 sql sql.rstrip().rstrip(;) LIMIT 100 return {passed: True, sql: sql}拦截掉危险SQL的功劳主要在这一层。出现过一次模型把“近30天退款订单数”理解成对退款表做DROP的极端情况——校验层直接拦下来回去重生成避免了风险。4.5 会话记忆如何在多轮对话中维持查询上下文问数场景天然是多轮对话用户第一轮问“上个月各品类GMV”第二轮接着说“只看华东区”你要能理解“只”字是在第一轮的结果上做条件过滤。这要求Agent有会话记忆能力。我们的实现是用Redis存最近N轮对话摘要和最近一轮的SQL执行结果。具体做法是每轮结束后把用户问题和最终生成的SQL压缩成一个结构化记录存进Redis。下一轮追问进来取出历史记录拼接到当前Prompt的上下文位置。注意拼接的是“历史问题对应SQL”而不是把所有历史原文全塞进去这样既保留了执行上下文又不至于Token爆炸。会话记忆还有一个容易被忽略的作用——结果对比。用户问“上周”之后继续问“那这周呢”你需要在结果解释节点里拿到上周的结果跟本周做对比分析。所以State的ResultState里始终保留上轮执行结果的摘要这个设计在真实使用中好评度很高。5. 架构演进与后续规划这套架构上线跑了两周整体稳定性比预期好但暴露出的问题也值得记录。第一个问题是Schema召回在复杂查询时不够精准比如用户问“高价值用户”系统只召回“用户”相关表但真正有价值的信息分散在“订单”和“用户等级”两张表里需要多表关联。这个问题单靠向量检索很难解决我们的方案是提前在知识层维护“业务主题-多表关联关系”的映射做一次业务层面的预聚合建模。第二个问题是长任务的处理。当前架构是同步请求-响应模式用户等几秒钟能拿到结果。但有些复杂查询比如跨月、跨表、大聚合可能要跑几十秒同步等待的体验就很差。后续规划是把任务队列和异步通知机制引入复杂查询先提交任务完成后通过企微机器人推送结果。第三个方向是MCP协议的全面适配。我们已经按MCP的思路设计了工具层后续等生态成熟可以直接切换。接入更多数据源ClickHouse、Elasticsearch时工具层只要能注册对应执行器就行架构上不用做大改动。再往后我们希望加入Agent的自我学习能力——把每轮“校验失败-重试成功”的案例自动沉淀进Few-shot库让系统越用越聪明。这个在评估安全性比如防止用户通过SQL注入给系统投毒之后才会动工。问数Agent的架构设计本质上是在“模型能力边界”和“业务对准确性的要求”之间找平衡。模型再强也会偶尔出错但通过合理的架构设计——拆节点、加校验、做召回、带记忆——能把出错率压到可以接受的范围。这也是我想通过这系列文章传达的核心思路架构不是炫技而是拿来对抗模型不确定性的。
