我最近在带一个内部项目，代号叫LCODER，核心是给业务团队做一个“问数”智能体——让不懂SQL的运营、产品同学直接用自然语言查数。目前第一版已经跑通了端到端流程，这里把项目架构的设计过程、技术选型和落地踩坑记录下来。如果你正打算从零搭一个AI Agent项目，或者在公司做Text-to-SQL类的智能体，这篇应该能帮你少走一些弯路。
先说清楚LCODER问数项目到底解决什么问题。传统的取数流程是业务提需求、数据团队写SQL、排期交付，一个简单查询动辄等半天。我们的目标是让业务方在对话框里直接问“上周华东区各品类GMV环比变化”，智能体自动完成语义理解、表字段匹配、SQL生成、结果执行和可视化返回，把整个链路压缩到几秒钟。听起来挺美好，但真正落地的时候会发现，这不只是“接个大模型API”那么简单——中间涉及多Agent协作、工具调用、知识库召回、会话记忆等一堆设计决策。这篇文章是系列第一篇，聚焦在项目架构层，讲清楚我们为什么这么拆、每个模块的职责边界，以及实际编码时的关键实现。
1. 整体设计思路：为什么问数场景必须用多Agent而不是单Prompt
刚开始设计的时候，我也试过“一步到位”——一个大模型调用，把数据库Schema、用户问题、few-shot示例全塞进Prompt里，让模型直接输出SQL。简单场景下（比如两三张表、几十个字段）确实能跑，但一旦业务库表复杂起来，问题马上暴露。
1.1 单Prompt方案在复杂问数场景下的三个致命问题
第一个问题是上下文爆炸。我们的核心库有几十张表、上千个字段，如果全部塞进Prompt，Token数量轻松破万。这不仅浪费成本，更重要的是模型在面对超长上下文时注意力会分散，反而更不容易抓住关键信息。实测下来，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协议标准的示例风格，同时自研轻量工具注册框架。MCP（Model 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_factory=list) recalled_columns: list = Field(default_factory=list) 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_factory=SchemaState) sql: SQLState = Field(default_factory=SQLState) result: ResultState = Field(default_factory=ResultState) meta: dict = Field(default_factory=dict)接着把五个节点函数装配成图。每个节点函数签名都是(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的架构设计，本质上是在“模型能力边界”和“业务对准确性的要求”之间找平衡。模型再强也会偶尔出错，但通过合理的架构设计——拆节点、加校验、做召回、带记忆——能把出错率压到可以接受的范围。这也是我想通过这系列文章传达的核心思路，架构不是炫技，而是拿来对抗模型不确定性的。
