1. 项目全景与基础设施需求拆解1.1 问数智能体到底是做什么的我先把这个项目的定位说清楚。LCODER这个系列的第一篇讲的是方案设计到了第二篇就是动真格的时候了。所谓的“问数项目”准确说是做一 个Text-to-SQL形态的智能体用户用一句大白话提问比如“上个月华东区各品类的销售额排名是什么样的”智能体负责理解意图、匹配到正确的表和字段、生成SQL、到数据库里把数据查出来最后用自然语言把结果解释给用户听。这句话读起来轻飘飘的但拆开看就有很多硬骨头自然语言怎么转成结构化查询大模型幻觉导致SQL写错怎么办数据库表太多召回不准怎么办查询结果怎么判断要不要展示给用户……这些问题的答案恰恰都依赖一个设计合理的基础设施层。你可以把基础设施理解为房子的地基和管线——用户看不见但如果没铺好后续每一层业务功能都会在各种莫名其妙的地方漏水。这一篇把基础设施搭建讲透目的在于把后续开发工作里的不确定性提前消化掉。如果你也准备做一个类似的企业级问数机器人这一篇可以直接跟着复现如果你只是对Agent开发感兴趣这些基础设施选型思路和踩坑过程同样能帮你避免“模型对话没问题、一接数据库就崩”的尴尬。1.2 基础设施要做哪几件事我从实际需要出发把基础设施拆成四个层面这个分类也直接对应后面的章节安排。第一层是工程基座Python环境、依赖管理、目录结构、配置管理、日志系统。这些不起眼的东西决定了团队协作时会不会互相踩脚也决定了代码半年后还能不能继续维护。第二层是模型接入层大模型API怎么统一封装、密钥怎么管理、超时重试怎么做、流式输出怎么处理。这是所有Agent能力的“水电煤”同时也往往是团队最容易乱搞的一层。第三层是数据设施层业务数据库连接、表结构元数据的管理、向量数据库的初始化、Embedding模型接入。问数项目的数据链路是全局最复杂的一段牵涉结构化数据和非结构化索引的配合必须在一开始就打好底。第四层是运行时设施核心Agent框架的初始化、工具注册机制、对话记忆、外部服务化API以及缓存与日志追踪。这层决定了对外的服务能力和后续迭代的效率。1.3 技术选型总览我把整个基础设施的选型方案先列出来后面的章节会逐个解释为什么选它层级组件选型替代方案开发语言Python3.113.10也可但不建议低于3.10依赖管理uvPoetry / pip-toolsAgent框架LangChain 0.3.xLlamaIndex轻量场景可手写Web服务FastAPIFlask / Quart业务数据库PostgreSQL 15MySQL 8.0向量数据库Qdrant 1.9Milvus / pgvector缓存/会话Redis 7KeyDB大模型接入OpenAI兼容协议各家原生SDK这套方案不是拍脑袋选的。比如向量库为什么不用pgvector——如果业务库本身有地理信息、全文检索之外还需处理大规模向量pgvector的性能和运维隔离感都差点意思Qdrant单机部署足够轻后期需要分布式扩展也有成熟的方案。用PostgreSQL存业务数据用Qdrant存表结构描述和样例数据的向量索引两者各管一摊界限清晰。2. 工程基座与项目脚手架搭建2.1 Python环境与依赖管理问数智能体涉及的技术栈非常杂LangChain、数据库驱动、向量库客户端、Web框架、数据处理库每个库又有自己的依赖树很容易出现依赖冲突。所以环境的隔离和依赖锁定必须从第一天就做好不然后面装一个包就崩一次环境心态会直接报废。我用的是Python 3.11加uv。uv是目前Python生态里速度最让人舒适的包管理器底层用Rust写的装依赖比pip快上好几倍同时生成的lock文件能保证团队所有人的环境完全一致。如果你还在用Python 3.9或者3.10建议至少升到3.10因为Type Hint的语法支持会舒服很多Agent代码里大量用到类型注解老版本写起来很憋屈。初始化项目的步骤很简单# 安装uvmacOS/Linux curl -LsSf https://astral.sh/uv/install.sh | sh # 初始化项目并创建虚拟环境 uv init lcoder-agent cd lcoder-agent uv venv --python 3.11 .venv source .venv/bin/activate # 安装核心依赖 uv add langchain langchain-openai langchain-community uv add fastapi uvicorn[standard] pydantic-settings uv add psycopg[binary] sqlalchemy uv add qdrant-client uv add redis uv add pandas openpyxl uv add loguru注意这里我把pandas和openpyxl也加进来了。问数Agent查完数据之后往往需要做一些聚合、排序、格式转换再把结果渲染成表格或图表数据pandas是这类工作的主力工具。openpyxl是为后续直接导出Excel报表预留的如果项目没有这个需求可以去掉。2.2 项目目录结构设计基础设施阶段最重要的产出之一就是一套清晰的目录结构。我见过太多Agent项目所有代码都塞在两三个文件里刚开始跑demo很爽一旦要加工具、加技能、加知识库整个项目就变成一团乱麻。我这边的目录设计如下lcoder-agent/ ├── app/ │ ├── main.py # FastAPI入口 │ ├── config.py # 配置管理 │ ├── api/ # HTTP接口层 │ ├── agent/ # Agent核心 │ │ ├── agent.py # 主Agent封装 │ │ ├── prompts.py # 提示词模板 │ │ └── skills/ # 工具/技能注册 │ ├── tools/ # 具体工具实现 │ │ ├── sql_executor.py # SQL执行工具 │ │ └── db_metadata.py # 元数据查询工具 │ ├── models/ # 数据模型 │ ├── services/ # 业务服务 │ ├── vector/ # 向量库相关 │ └── utils/ # 工具函数 ├── data/ # 本地数据文件 ├── logs/ # 日志目录 ├── scripts/ # 初始化脚本 ├── tests/ # 测试 ├── .env.example # 环境变量模板 ├── pyproject.toml └── README.md这套结构的关键在于把“Agent逻辑”和“工具实现”做了隔离。Agent层只负责调度——决定下一步调哪个工具、怎么处理工具的返回结果而具体的工具实现放在tools目录比如SQL执行器、元数据查询器。这样后续加一个新工具不需要改Agent的主逻辑只要在skills里注册一下就行扩展性很好。2.3 配置管理与环境变量Agent项目里有大量需要外部注入的配置项比如大模型的API密钥、数据库连接字符串、向量库地址、Redis地址。我强烈建议所有配置走环境变量并且用一个config.py统一读取和管理。我是用pydantic-settings来做这件事的它最大的好处是能在启动时做类型检查和必填校验避免密钥或地址配错时到了使用环节才报错。# app/config.py from functools import lru_cache from pydantic_settings import BaseSettings, SettingsConfigDict class Settings(BaseSettings): model_config SettingsConfigDict( env_file.env, env_file_encodingutf-8, extraignore ) # 大模型配置 LLM_MODEL: str gpt-4o-mini LLM_API_BASE: str LLM_API_KEY: str LLM_TEMPERATURE: float 0.1 LLM_MAX_TOKENS: int 4096 LLM_TIMEOUT: int 60 # Embedding配置 EMBEDDING_MODEL: str text-embedding-3-small EMBEDDING_API_BASE: str EMBEDDING_API_KEY: str # 业务数据库配置 DB_HOST: str localhost DB_PORT: int 5432 DB_USER: str postgres DB_PASSWORD: str DB_NAME: str business_db DB_POOL_SIZE: int 10 # 向量库配置 QDRANT_HOST: str localhost QDRANT_PORT: int 6333 QDRANT_COLLECTION: str db_metadata # Redis配置 REDIS_HOST: str localhost REDIS_PORT: int 6379 REDIS_DB: int 0 REDIS_PASSWORD: str # 服务配置 API_HOST: str 0.0.0.0 API_PORT: int 8000 lru_cache def get_settings(): return Settings()注意两个细节。第一LLM_TEMPERATURE我默认调到0.1问数场景需要的是准确不是创意温度越高SQL写错的概率越大。第二用lru_cache装饰get_settings确保整个应用生命周期里配置只加载一次避免反复读取.env文件浪费IO。对应的.env.example模板是这样LLM_MODELgpt-4o-mini LLM_API_BASEhttps://your-llm-endpoint.example.com/v1 LLM_API_KEYsk-your-key-here DB_HOSTlocalhost DB_PASSWORDchange-me QDRANT_HOSTlocalhost REDIS_HOSTlocalhost.env文件绝对不能提交到Git仓库这条要写进.gitignore一旦密钥泄露到代码仓库后续换密钥的成本比任何人想象的都高。2.4 日志与调试设施基础设施阶段必须把日志系统铺好否则等到Agent跑起来之后LLM的输入输出、工具的调用链、SQL的执行情况全都没有记录出了问题你根本不知道是哪个环节挂了全靠猜是非常痛苦的。我之前用Python自带的logging后来换成loguru它的使用体验更顺手配置简单日志格式也更直观。我的做法是按天切分日志文件并同时输出到控制台和文件方便开发时实时看和事后排查。# app/utils/logger.py import sys from loguru import logger logger.remove() logger.add( sys.stdout, levelINFO, formatgreen{time:YYYY-MM-DD HH:mm:ss}/green | level{level: 8}/level | cyan{name}/cyan:cyan{line}/cyan - level{message}/level ) logger.add( logs/agent_{time:YYYY-MM-DD}.log, levelDEBUG, rotation00:00, retention30 days, encodingutf-8, enqueueTrue, format{time:YYYY-MM-DD HH:mm:ss} | {level: 8} | {name}:{line} | {message} ) core_logger logger这里retention设30天保证能追踪一个月内的问题轨迹。enqueueTrue是让日志写入走异步队列避免在高并发环境下日志I/O阻塞主业务逻辑。3. 大模型接入层搭建3.1 统一走OpenAI兼容协议问数Agent的大模型接入有一个很关键的决策不要绑死某一家厂商的SDK。现在各家大模型厂商都提供了OpenAI兼容的接口格式这意味着你可以用同一个OpenAI客户端库去对接几乎所有主流模型服务包括各类国产模型、开源模型的私有化部署服务。实际做法是在配置里指定LLM_API_BASE指向对应厂商的地址然后统一使用langchain-openai提供的ChatOpenAI类来实例化模型。这个做法的好处是后续换模型时只改配置不改代码对模型做A/B对比测试时更省事。# app/agent/llm.py from langchain_openai import ChatOpenAI from functools import lru_cache from app.config import get_settings DEFAULT_SYSTEM_PROMPT 你是一名数据分析助手负责根据用户的问题编写并执行SQL查询来回答业务问题。请遵循以下原则 1. 只使用提供的数据库表结构信息不要臆测不存在的字段。 2. SQL查询结果为空时如实告知用户不要编造数据。 3. 涉及金额时保留两位小数。 4. 严格按照工具返回的结果组织回答。 lru_cache def get_llm(): settings get_settings() return ChatOpenAI( modelsettings.LLM_MODEL, api_keysettings.LLM_API_KEY, base_urlsettings.LLM_API_BASE, temperaturesettings.LLM_TEMPERATURE, max_tokenssettings.LLM_MAX_TOKENS, timeoutsettings.LLM_TIMEOUT, max_retries2, )3.2 密钥管理与安全边界模型密钥的安全级别和数据库密码是同一个等级这是基础设施阶段最容易忽视的问题。我见过不止一个团队把API密钥直接写在代码里甚至提交到Git仓库结果被扫描工具扫出来之后只能紧急轮换密钥。我的建议是至少做到三点。第一密钥只存在于.env文件且.env不入库仓库里只保留.env.example。第二团队内部用环境变量注入的方式分发密钥不要通过聊天工具传明文。第三后端服务调用模型前端浏览器永远接触不到模型密钥。如果你的Agent对外提供服务一定不要做成前端直连大模型API的架构中间必须隔一层后端代理否则密钥会在浏览器开发者工具里暴露无遗。3.3 模型调用的可靠性设计大模型API在生产环境里并不像本地跑demo那么稳定超时、限流、网络抖动都是常态。问数项目中模型调用是链路最上层的环节一旦模型接口超时或返回异常下游SQL执行和结果展示全都跟着失败所以必须在接入层就把可靠性机制做扎实。第一件要做的是超时控制。ChatOpenAI的timeout参数一定要设我之前用的默认值经常在模型服务负载高时把请求挂到两分钟以上用户那边早就等得不耐烦了。目前我这边设置60秒对普通非流式请求来说60秒足够如果超过这个时间基本可以判定模型服务异常继续等待没有意义。第二件是重试机制。max_retries设2次配合OpenAI客户端的内置重试逻辑可以在请求失败时自动重试避免因为偶发的网络问题直接返回错误给用户。第三件是请求日志。每次模型调用的入参和出参都要记录但这里有个安全提醒日志里不要记录完整的API Key同时考虑是否需要对SQL查询结果做脱敏这取决于你的数据敏感程度。从问数项目长远来看这个数据权限问题会在后文中详细展开基础设施阶段先留好日志扩展位即可。3.4 Embedding模型与向量化准备除对话模型外基础设施阶段还要把Embedding模型接入好。问数Agent在做表结构召回时需要把用户的问题和表的描述信息都转成向量然后在向量库里做相似度检索这就要用到Embedding模型。Embedding模型的接入方式和对话模型类似用langchain-openai里的OpenAIEmbeddings# app/vector/embedding.py from langchain_openai import OpenAIEmbeddings from functools import lru_cache from app.config import get_settings lru_cache def get_embeddings(): settings get_settings() return OpenAIEmbeddings( modelsettings.EMBEDDING_MODEL, api_keysettings.EMBEDDING_API_KEY, base_urlsettings.EMBEDDING_API_BASE, timeout30, max_retries2, )Embedding模型的选择有几个考虑维度。第一是维度大小常见的有1536维也有1024或768维的维度越高精度通常越好但存储和计算成本也越高。第二是中文效果一定要拿实际表结构描述去做评测不同模型在中文业务文本上的效果差异可能超出你的预期。第三是调用成本Embedding是批量调用的元数据初始化时要对大量文本做向量化量大的时候成本不可忽略选型时要算清楚账。4. 数据层与向量检索设施搭建4.1 业务数据库连接管理问数项目的核心操作对象就是业务数据库这层的连接管理质量直接决定了Agent查询数据的稳定性和速度。PostgreSQL是我最常用的选择在复杂SQL支持、JSON处理、扩展生态方面都很成熟用它存业务数据非常稳妥。SQLAlchemy是Python里最主流的数据访问层它提供了连接池管理、ORM映射、SQL表达式语言等能力。在Agent场景下我们既要执行动态生成的SQL也需要ORM层来做一些配置管理和元数据记录两者不冲突。# app/services/database.py from sqlalchemy import create_engine, text from sqlalchemy.pool import QueuePool from contextlib import contextmanager from functools import lru_cache from app.config import get_settings lru_cache def get_engine(): settings get_settings() db_url ( fpostgresqlpsycopg://{settings.DB_USER}:{settings.DB_PASSWORD} f{settings.DB_HOST}:{settings.DB_PORT}/{settings.DB_NAME} ) return create_engine( db_url, poolclassQueuePool, pool_sizesettings.DB_POOL_SIZE, max_overflow5, pool_pre_pingTrue, pool_recycle1800, ) contextmanager def get_db_connection(): engine get_engine() conn engine.connect() try: yield conn finally: conn.close() def execute_sql(sql: str): 执行只读查询返回结果集 with get_db_connection() as conn: result conn.execute(text(sql)) columns list(result.keys()) rows [dict(zip(columns, row)) for row in result.fetchall()] return {columns: columns, rows: rows, row_count: len(rows)}几个关键参数我要重点解释一下。pool_pre_pingTrue是在每次从连接池取连接之前先发送一个ping信号检测连接是否存活避免数据库端连接因为空闲太久被断开拿到一个坏连接导致查询报错。pool_recycle1800表示连接最多存活30分钟就要回收重建防止数据库端主动断连造成的“connection already closed”错误。max_overflow5意味着连接池在繁忙时最多可以额外创建5个临时连接超过这个数就排队等待——问数场景不需要盲目的高并发限制连接数反而是对数据库的一种保护。4.2 只读账号与权限的最小化设计这件事我必须放在数据库连接这一节里重点强调问数Agent连数据库绝对不能使用业务系统管理员的账号。原因很简单Agent再智能本质上是代码在执行SQL代码一旦出现逻辑漏洞或者被提示词注入攻击就可能执行出删表、改数据这类破坏性操作。我在项目里单独创建了一个只读账号CREATE USER agent_readonly WITH PASSWORD strong-password; GRANT CONNECT ON DATABASE business_db TO agent_readonly; GRANT USAGE ON SCHEMA public TO agent_readonly; GRANT SELECT ON ALL TABLES IN SCHEMA public TO agent_readonly; ALTER DEFAULT PRIVILEGES IN SCHEMA public GRANT SELECT ON TABLES TO agent_readonly;ALTER DEFAULT PRIVILEGES这一行很多人容易忽略。它保证了后续新建的表默认给只读账号授予SELECT权限否则表建好后Agent又查不了排查半天才发现是权限问题。除了数据库侧应用侧也要做一个兜底防线。SQL执行器里检查SQL语句强制执行语句前判断第一条非注释语句的动词必须是SELECT、WITH、SHOW、EXPLAIN这些只读操作其他的直接拒绝执行。虽然模型生成的SQL绝大多数情况不会写DELETE但安全边界永远要相信自己写的代码不要相信外部模型的输出。4.3 表结构元数据提取与加工问数Agent要写好SQL前提是它得知道数据库里有哪些表、每张表有哪些字段、字段的业务含义是什么。这些信息就是数据库的“元数据”。元数据可以从PostgreSQL的系统目录直接查询SELECT t.table_name, c.column_name, c.data_type, c.is_nullable FROM information_schema.tables t JOIN information_schema.columns c ON t.table_name c.table_name WHERE t.table_schema public AND t.table_type BASE TABLE ORDER BY t.table_name, c.ordinal_position;但只靠系统目录还不够。系统目录里的表和字段名往往是英文缩写或者技术命名比如cst_amt模型根本猜不出这是“客户金额”的意思。所以我额外做了一个“字段注释”的加工环节把业务注释写入PostgreSQL的COMMENT里COMMENT ON TABLE sales_order IS 销售订单表每行代表一条客户订单记录; COMMENT ON COLUMN sales_order.cst_amt IS 客户订单总金额元含税; COMMENT ON COLUMN sales_order.order_status IS 订单状态0-草稿1-已提交2-已发货3-已完成4-已取消;这些COMMENT信息同样可以从系统目录读出来我们在做元数据时要把“表名字段名字段类型业务注释”组合成一段完整描述文本后续它有两个用途一是直接作为提示词上下文喂给大模型帮助它理解表结构二是做向量化后存入向量库用于检索召回。4.4 向量数据库的初始化与集合管理Qdrant的初始化非常轻量本地用Docker启动一个单节点就够开发调试了docker run -d --name qdrant \ -p 6333:6333 -p 6334:6334 \ -v $(pwd)/qdrant_storage:/qdrant/storage \ qdrant/qdrant:v1.9.7注意6334端口是gRPC接口生产环境建议用内部网络通gRPC提升性能开发阶段用6333的HTTP接口就够。集合的创建考虑两个核心参数向量维度和距离度量方式。向量维度必须和你选的Embedding模型输出维度一致比如text-embedding-3-small是1536维如果后续换模型向量维度变了旧集合的数据就完全失效了。距离度量我用Cosine尤其在文本相似度场景下Cosine比Dot Product更直观稳定对向量的模长不敏感适合描述文本语义的接近程度。# scripts/init_qdrant.py from qdrant_client import QdrantClient from qdrant_client.http import models as qdrant_models client QdrantClient(hostlocalhost, port6333) client.recreate_collection( collection_namedb_metadata, vectors_configqdrant_models.VectorParams( size1536, distanceqdrant_models.Distance.COSINE, ), ) print(collection created)在采样数据上我再多提一句。问数Agent的召回策略里表结构文本做向量化后存入向量库而表名和字段名可以同时做关键词过滤或全文匹配。在实际使用时两者结合的效果比单用向量好很多因为表名的语义往往是精准匹配更可靠。向量检索负责语义接近性关键词匹配负责精确对应这个双通道策略在第五章还会展开。5. 智能体运行框架搭建5.1 Agent框架选型LangChain还是LlamaIndex问数项目里关于框架的选型我的结论是主Agent编排用LangChain数据处理和检索部分参考LlamaIndex的设计思路但不直接引入整个LlamaIndex。为什么不用LlamaIndex做主线LlamaIndex在文档问答、知识库RAG场景下有巨大优势尤其是数据索引和数据加载器的生态非常完善。但问数项目的核心是Text-to-SQL加上动态工具调用LangChain在这块的抽象更成熟AgentExecutor和工具装饰器用起来更顺手社区案例也更多。为什么不完全手写如果你只是调一个模型、连一个数据库手写没问题。但问数Agent后续要加记忆、加多工具协同、加复杂的编排逻辑LangChain提供了这些组件的基础抽象把Web搜索、API调用、代码执行等工具接入标准化省掉了重复造轮子的时间。5.2 工具注册与技能机制这一节是整个Agent基础设施里最核心的部分对应目前很多团队在实践的“Agent Skill”机制。我的工具注册设计很简单用装饰器把函数注册到工具中心注册时附带名称、描述、参数SchemaAgent运行时通过读取注册中心来决定调用哪个工具。# app/tools/registry.py from typing import Callable, Dict, Any from pydantic import BaseModel, create_model class ToolSpec(BaseModel): name: str description: str args_schema: Any func: Callable _tool_registry: Dict[str, ToolSpec] {} def register_tool(name: str, description: str, args_schema: type[BaseModel]): def decorator(func): _tool_registry[name] ToolSpec( namename, descriptiondescription, args_schemaargs_schema, funcfunc, ) return func return decorator def get_all_tools(): return list(_tool_registry.values())工具描述文本的写法对Agent的效果影响比想象中大得多。描述里要写清楚这个工具是做什么的、在什么条件下使用、参数怎么填、有什么限制。大模型是靠描述来决定调用哪个工具的描述写得含糊它就会在你预期的工具之外瞎猜。这个设计最大的好处是模块解耦。每加一个新的数据查询能力比如查ERP系统的接口、查报表平台的API只要实现一个函数并注册Agent的编排逻辑完全不需要变。后面想加权限校验、审计日志也只需要在注册中心统一包一层不需要改动每个工具的内部实现。5.3 SQL执行工具的实现有了注册机制接下来实现问数项目里最核心的工具SQL执行器。它要接收大模型生成的SQL执行查询并返回结果。但这里有一个真实的工程问题是大模型生成的SQL很可能有语法错误也可能踩到数据库不支持的语法。# app/tools/sql_executor.py from pydantic import BaseModel, Field from app.tools.registry import register_tool from app.services.database import execute_sql from app.utils.logger import core_logger class SQLExecuteArgs(BaseModel): sql: str Field(description要执行的SQL查询语句必须是SELECT只读查询) register_tool(sql_executor, 执行只读SQL查询语句返回查询结果, SQLExecuteArgs) def sql_execute(sql: str): core_logger.info(f[SQL Executor] executing: {sql}) # 安全检查只允许以SELECT开头的语句 stripped sql.lstrip().lstrip(().strip().lower() if not stripped.startswith((select, with, show, explain)): return {error: 仅允许执行只读查询SELECT/WITH/SHOW/EXPLAIN} try: result execute_sql(sql) core_logger.info(f[SQL Executor] rows returned: {result[row_count]}) return result except Exception as e: core_logger.error(f[SQL Executor] error: {e}) return {error: str(e)}工具的返回格式很重要。不要把数据库原始结果直接扔给模型而是要做结构化的封装。模型拿到结果后还需要判断是否满足用户的原始问题如果不满足可能要调整SQL重查。所以返回结果最好加上列名、行数、以及前N行数据。数据量大时还得做截断这个后面会详细讲。5.4 元数据召回工具的实现问数Agent使用元数据工具的方式和人类分析师很相似。人分析数据之前得先看有哪些表、哪些字段Agent也是一样——先查到可用的表结构把它作为提示词的上下文才能进入SQL生成阶段。元数据工具有两条获取路径。一条是精准路径根据用户问题中出现的表名直接查系统目录拿表结构另一条是模糊路径把用户问题做向量化在Qdrant里做相似度检索把最相关的表描述召回出来。# app/tools/db_metadata.py from pydantic import BaseModel, Field from app.tools.registry import register_tool from app.services.metadata import get_table_schema class SearchSchemaArgs(BaseModel): keywords: str Field(description表名或字段名的关键词) register_tool(search_table_schema, 搜索业务数据库中与关键词相关的表和字段结构, SearchSchemaArgs) def search_table_schema(keywords: str): return get_table_schema(keywords)在实际项目中精准路径和模糊路径建议都实现。精准路径用来快速定位——当用户提到“订单表”时直接拿到sales_order的结构模糊路径用来兜底——用户用“客户买了多少钱”这种口语化描述时通过向量检索找到cst_amt字段。基础阶段先把两条路径都留好具体调参在后续的业务逻辑章节里继续做。5.5 主Agent运行时封装最后把模型、工具、记忆、提示词粘合在一起的就是主Agent本体。我用LangChain的create_react_agent来构建。ReAct模式的核心思想是让模型在“思考”和“行动”之间循环思考当前需要什么信息、决定调用哪个工具、观察工具的返回结果、继续思考下一步直到有足够信息回答用户。这种模式非常适合问数场景因为要回答“订单金额最大的客户是谁”模型需要调用元数据工具看表结构、调用SQL执行器查数据这中间至少要两个步骤。# app/agent/agent.py from langchain.agents import create_react_agent, AgentExecutor from langchain_core.prompts import PromptTemplate from langchain_core.tools import StructuredTool from app.agent.llm import get_llm, DEFAULT_SYSTEM_PROMPT from app.tools.registry import get_all_tools def build_agent(): llm get_llm() tools [] for spec in get_all_tools(): tools.append(StructuredTool.from_function( funcspec.func, namespec.name, descriptionspec.description, args_schemaspec.args_schema, )) prompt PromptTemplate.from_template(DEFAULT_SYSTEM_PROMPT) agent create_react_agent( llmllm, toolstools, promptprompt, stop_sequenceTrue, ) executor AgentExecutor( agentagent, toolstools, verboseTrue, handle_parsing_errorsTrue, max_iterations5, return_intermediate_stepsTrue, ) return executormax_iterations5是很重要的参数。它限制Agent最多只能思考和行动5轮超过就直接返回避免大模型陷入无限循环、反复调用工具停不下来。真实项目里一次查询很少超过4轮如果超过了往往是提示词或工具描述出了问题排查的时候优先看中间步骤的日志基本能定位到是哪一步的意图没被识别对。6. 服务化与工程化配套搭建6.1 FastAPI封装Agent服务Agent内核搭好了还需要把它封装成服务才能让前端页面或者同事的客户端调用。FastAPI在这层做得非常舒服类型提示天然支持、OpenAPI文档自动生成、异步性能也足够。# app/main.py from fastapi import FastAPI from pydantic import BaseModel from app.agent.agent import build_agent app FastAPI(titleLCODER问数智能体) class QueryRequest(BaseModel): question: str session_id: str history: list[dict] [] class QueryResponse(BaseModel): answer: str sql: str data: dict {} intermediate_steps: list [] agent_executor build_agent() app.post(/api/query, response_modelQueryResponse) async def query(req: QueryRequest): # 将结构化参数包装成给Agent的输入 result await asyncio.to_thread( agent_executor.invoke, {input: req.question} ) return QueryResponse( answerresult[output], intermediate_stepsresult.get(intermediate_steps, []) )这里有一个要点AgentExecutor的invoke方法是同步阻塞的直接放进FastAPI的async函数里会阻塞事件循环导致其他请求排队。用asyncio.to_thread把同步调用丢到线程池里执行不阻塞主事件循环实测在普通服务器上的并发能力就够用了。6.2 Redis缓存与轻量会话管理如果每次请求都重新走一次建模-推理-SQL生成的完整流程不仅慢而且模型成本会高得惊人。对于一些高频的、相似的查询可以加一层缓存。Redis在这里充当两个角色。第一是查询结果缓存把问题做规范化处理后作为key第一次查询把结果写入Redis并设置过期时间后续相同问题直接命中。第二是会话上下文缓存存储Agent和用户的对话记录支持多轮问答的场景。比如按用户question的哈希值做缓存import hashlib import json from redis import Redis from app.config import get_settings settings get_settings() redis_client Redis( hostsettings.REDIS_HOST, portsettings.REDIS_PORT, dbsettings.REDIS_DB, decode_responsesTrue, ) def get_cache_key(question: str) - str: digest hashlib.md5(question.strip().lower().encode()).hexdigest() return fagent:query:{digest} def set_query_cache(question: str, answer: dict, expire_seconds: int 3600): key get_cache_key(question) redis_client.setex(key, expire_seconds, json.dumps(answer, ensure_asciiFalse)) def get_query_cache(question: str): key get_cache_key(question) cached redis_client.get(key) return json.loads(cached) if cached else None缓存粒度是值得斟酌的。如果整个SQL查询结果都缓存数据变化后缓存会过期不准如果只缓存SQL生成结果每次执行SQL才能拿最新数据但大模型调用省了。折中方案是缓存的过期时间根据业务数据新鲜度来调整对实时性要求高的报表类型直接跳过缓存。这一块属于业务策略的范畴基础设施阶段先把缓存的通道打通后面按需调整就行。6.3 全链路可观测与排查辅助Agent项目的调试难度比普通Web服务高一截关键在于链路太长用户请求先进FastAPI然后进AgentExecutor再到大模型API模型返回后可能调元数据工具、再调SQL执行器最后拼接答案返回。任何一个环节出错表现在用户端的都只是“回答不对”或“超时”。所以基础设施阶段必须把可观测性搭好。不只是日志而是要把一次请求的关键节点串联起来。我目前的方案是在每个工具调用和模型调用前后打点import time import uuid from loguru import logger from contextvars import ContextVar request_id_var: ContextVar[str] ContextVar(request_id, default-) def log_with_context(message: str, **kwargs): rid request_id_var.get() logger.info(f[{rid}] {message}, **kwargs) def track_execution(func, name: str): def wrapper(*args, **kwargs): start time.monotonic() log_with_context(f{name} start, argsargs) try: result func(*args, **kwargs) elapsed time.monotonic() - start log_with_context(f{name} end, elapsed_msround(elapsed * 1000, 2)) return result except Exception as e: elapsed time.monotonic() - start log_with_context(f{name} error, elapsed_msround(elapsed * 1000, 2), errorstr(e)) raise return wrapper在FastAPI入口为每个请求生成一个request_id通过ContextVar贯穿整个请求生命周期。这样排查问题时只需要用一个request_id把日志文件里所有相关记录捞出来就能还原一次完整调用链效率比没有request_id时大海捞针高得多。7. 常见问题与排查技巧实录7.1 依赖冲突与版本地狱LangChain生态的版本更新速度非常快不同大版本之间API可能完全不兼容。我遇到的典型问题是项目里装了一个依赖库它依赖的是旧版langchain-core结果跟新装langchain-openai的版本冲突程序启动时报出一堆ABC签名错误。排查这类问题我的建议是按这几个步骤来第一启动时如果报错先别急着搜索报错信息检查itsdangerous的提示里有没有“conflict with”或者“requires a different version”的字样第二用pip list或uv pip list确认当前环境里langchain相关包的版本号第三锁定一个“全家桶版本组合”比如langchain 0.3.x配langchain-openai 0.2.x配langchain-community 0.3.x不要一个最新一个旧版混着装。还有一个小技巧uv的lock文件把版本精确锁定到commit级别团队协作时大家同步的是完全一致的环境。你自己调试时如果用pip自由装很容易在“我这边能跑啊”和“为什么我这边跑不了”之间反复横跳。一旦基础设施阶段锁好版本后面的各种集成问题会少非常多。7.2 数据库连接池被打满开发调试阶段可能没啥感觉一旦服务同时处理多个会话数据库连接池很容易被打满报错通常是“connection pool exhausted”或者“timeout waiting for connection”。这里要注意一个LangChain的隐藏行为默认情况下AgentExecutor每执行一步工具调用都会新建一个连接如果连接池没关好工具多轮调用就会把连接池耗尽。解决办法是确保execute_sql使用同一个engine connection池engine用lru_cache缓存工具执行完连接正常释放回到池里。还有一个排查姿势查看数据库端的pg_stat_activity表看是不是有大量idle in transaction状态的连接堆积。如果有多半是事务没有正常提交或回滚。检查代码里有没有把数据写入和读取操作混在同一个长事务里问数Agent只做读操作事务越短越好。SELECT pid, state, state_change, query_start, left(query, 80) AS query FROM pg_stat_activity WHERE datname business_db ORDER BY query_start;7.3 模型生成的SQL质量不稳定SQL生成质量是最让人头疼也最需要长期迭代的问题。基础设施阶段能做的准备是预设好错误反馈机制一旦SQL执行报错把报错信息返回给模型让它根据报错信息修正SQL重写。Agent工具的返回本身就是模型下一步决策的输入所以SQL执行器返回的错误信息要足够详细。PostgreSQL的报错信息对程序不太友好但对大模型其实是很好的修正信号——比如“column sales_order.cst_amt does not exist”这句话模型读完就能理解自己用错了字段名。在此基础上我还发现一个很有用的做法在元数据工具里把字段的枚举值或者数据分布概要也带上。比如订单状态字段如果明确告诉模型“order_status的取值是0草稿/1已提交/2已发货/3已完成/4已取消”模型生成带过滤条件的SQL时基本不会写错。这个信息可以通过查询字段的distinct value来获取但注意数据量大的字段不要做全量distinct会拖慢元数据工具性能。7.4 上下文令牌超限怎么办问数Agent的上下文里要放表结构描述、历史对话、工具调用结果内容很容易超过模型的token上限。尤其是元数据召回的数量控制不好时一次性塞给模型几十张表的描述光表结构就把上下文撑爆了模型反而抓不住重点。我的经验是一个查询流程里召回的元数据控制在3到5张表以内。召回更多时模型通常不会认真看反而增加token消耗和推理延迟。如果确实需要跨很多表查询先把问题拆解成子问题让Agent分步处理而不是一次性把所有表都塞进去。工具返回的结果集也要做截断。SQL查询如果返回了几千行数据不能一股脑扔给大模型生成回答那会让token瞬间爆炸。我的方案是最多返回前50行并附上“共有N行”的提示。如果模型需要更详细的数据它可以通过追加LIMIT子句或加GROUP BY聚合来重新查询。7.5 调试链路“中间步骤”是唯一真相Agent项目调试时最大的错觉就是模型回答错了但你看不出为什么错。FastAPI的接口返回里我把intermediate_steps字段也带上了这个字段记录的是Agent每一步的思考、调用的工具、工具的输入输出。排查问题的时候先看这个字段而不是猜。比如用户问“订单最多的前十个客户”结果回答错了。看中间步骤就能定位是工具没调对比如搜索表结构时没有命中客户表还是SQL写错了比如没加GROUP BY还是返回结果被截断了。这比让开发人员一遍遍重放问题高效太多。我在loguru里专门对intermediate_steps做了结构化输出用JSON格式记录。后续如果接入Lunary或LangSmith这类可观测性平台这些数据也能直接对接为模型调优和回归测试提供数据基础。从基础设施搭建的角度回顾这一阶段我把工程基座、模型接入、数据设施、运行框架、服务化与可观测性这几层都立住了。技术选型不一定是最新最潮的但每一层都是经过实际项目和踩坑验证过的组合。我个人体会是基础设施阶段花的时间完全值得因为后面每写一行业务代码都有稳的地基接着而如果地基没打好加功能的同时还得修补之前的破洞那种感觉才是真的折磨。下一步就可以在这个底座上填充具体的业务技能了比如问数SQL生成、结果校验、数据可视化这些模块都有得聊。
