先说个真实感受问数智能体这类项目AI Agent 的“智能”看起来在提示词和模型能力上但真正决定它能不能上线、能不能抗住业务方天天问数的全是基础设施。模型答非所问可以调prompt但环境装不上、数据库连不上、会话状态丢、工具调用没日志这些问题会把整个项目拖垮。这篇就是把我搭建“LCODER问数智能体”时基础设施部分的做法完整拆出来从技术选型到环境初始化从模型封装到向量库接入再到数据源管理和调试排错全部是可直接参考的落地经验。这篇适合正在做 AI Agent 开发、或者准备从 Demo 往生产环境推的工程师。如果你是刚接触智能体开发也能照着把基座搭起来后面再往里面加业务逻辑会顺很多。项目代号叫 LCODER定位明确让业务同事用自然语言直接问数据库Agent 负责理解问题、拆解任务、写 SQL、跑数、再把结果整理成能看的图表和文字结论。1. 项目整体设计与技术选型思路1.1 问数智能体的核心需求拆解在动手写代码之前先把需求拆明白。问数项目看起来简单——用户问一句返回一个数。但拆开之后会发现它至少包含五个环节理解用户的自然语言问题识别查询意图和时间范围等关键条件找到相关的数据表、字段和业务口径这一步通常依赖数据字典和元数据检索生成符合目标数据库方言的 SQL并且保证语义正确执行 SQL 并处理返回结果这里涉及权限控制、超时保护和结果集大小限制把结果组织成业务人员能看懂的结论必要时给出图表这五个环节对应到 AI Agent 里就是一个典型的多工具调用工作流。如果只用单次大模型调用来完成整个链路效果会非常不稳定因为“写 SQL”和“查表结构”本质是两件事混在一起模型容易顾此失彼。所以我的选择是用 LangGraph 来做 Agent 编排把流程拆成节点每个节点只做一件事节点之间通过状态对象传递信息。1.2 为什么选择 LangGraph 而非 Dify 或自研框架最近 AI Agent 开发的热度很高市面上的框架也确实多。Dify 这类智能体平台我试过最大的优势是上手快拖拽节点就能出一个原型非常适合给业务方演示。但问题也很明显一旦涉及复杂条件分支、自定义代码节点、精细化的 token 成本控制低代码平台就有点力不从心。LangGraph 给我最大的感受是把“状态机”的思想带进了 Agent 开发。它把每次对话的执行过程建模为一个图节点是处理逻辑边是流转条件状态对象在节点之间传递。这有三个明显好处流程可控。每个节点都可以单独调试出问题时能明确知道卡在哪一步状态可查。中间过程全部保留在状态对象里方便追踪和排错支持循环和条件分支。比如 SQL 执行报错之后可以让 Agent 根据错误信息重新生成这个能力在问数场景里几乎是刚需有人可能会问为什么不用 Spring AI Multi Agent 或者自己写一套如果你所在团队是 Java 技术栈Spring AI 确实值得考虑。但 LCODER 的核心是数据处理和快速迭代Python 生态在这一块更顺手LangGraph 社区活跃度也高踩坑能搜到解决方案。至于自研框架我劝大家除非有特殊的性能要求否则没必要重复造轮子。Agent 编排的难点不在“调用模型”而在“管理状态和处理异常”这些 LangGraph 已经帮我们解决了大部分。1.3 基础设施模块划分与部署架构基础设施搭建不是单指“装个依赖”而是把 Agent 运行所需的支撑模块全部就位。我划分了六个模块模块职责技术选型配置中心管理环境变量、模型参数、数据源连接信息pydantic-settings .env模型接入层统一封装大模型和嵌入模型调用支持重试和超时OpenAI SDK / 兼容接口Agent 编排管理节点流转和状态传递LangGraph数据源层连接业务数据库提供元数据查询和 SQL 执行能力SQLAlchemy 只读账号向量存储存放数据字典、FAQ、历史查询经验支持语义检索Chroma开发/ pgvector生产会话与缓存保存多轮对话状态、查询缓存结果Redis可观测性日志、链路追踪、token 统计loguru Langfuse依赖和组件数量不算多但每个模块的初始化细节都有讲究。比如数据库连接池的大小设置、超时时间、向量库的 collection 命名规范、Redis key 的过期策略这些如果不在一开始就定好规则后面项目越写越大光是改这些参数就够头疼的。2. 开发环境准备与工程初始化2.1 虚拟环境与 Python 版本选择问数项目用到的 LangGraph、Chroma 等库对 Python 版本有要求我直接选了 Python 3.11。3.10 也能跑但 3.12 有些依赖还没完全跟上没必要冒险。建议所有开发成员统一版本用 pyenv 管理避免出现“我本地能跑你那边报错”的尴尬。虚拟环境我用的是 venv简单直接。如果你用 conda 或者 poetry 也没问题关键是团队内部统一。LCODER 工程的依赖文件我拆成了 requirements-base.txt 和 requirements-dev.txt前者放运行依赖后者放 pytest、ruff 这类开发工具。python -m venv .venv source .venv/bin/activate pip install -U pip setuptools wheel pip install -r requirements-dev.txt2.2 核心依赖清单与版本锁定把核心依赖列一下这些是我实测过相对稳定的组合langgraph0.2.0 langchain0.3.0 langchain-openai0.2.0 openai1.40.0 fastapi0.115.0 uvicorn[standard]0.30.0 sqlalchemy2.0.30 psycopg2-binary2.9.9 redis5.0.0 chromadb0.5.0 pydantic2.8.0 pydantic-settings2.4.0 loguru0.7.2 tenacity8.4.0这里要特别提醒不要直接pip install langchain一把梭。langchain 这个包已经拆成了很多子包装全量包会有大量用不到的依赖安装时间长不说还容易和其他库冲突。我只需要 langchain-core、langchain-openai 和 langgraph 这三个核心包。依赖装完之后建议立刻pip freeze requirements-lock.txt锁定版本。AI 项目依赖更新太快今天装的版本明天可能就 breaking change锁定版本能保证团队成员和生产环境一致。2.3 环境变量与配置管理配置管理我用了 pydantic-settings原因很简单它支持类型校验和自动补全配置写错能在启动时立刻发现而不是运行到一半才报错。配置项我放在 .env 文件里同时维护一个 .env.example 作为模板提交到代码仓库。# app/config.py 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_base_url: str llm_api_key: str embedding_model: str text-embedding-3-small temperature: float 0.2 max_tokens: int 2048 # 数据库配置 db_host: str localhost db_port: int 5432 db_user: str db_password: str db_name: str # Redis 配置 redis_url: str redis://localhost:6379/0 # 向量库配置 vector_store_path: str ./data/chroma collection_name: str lcoder_metadata # Agent 配置 max_iterations: int 5 sql_timeout: int 10 property def db_url(self) - str: return fpostgresqlpsycopg2://{self.db_user}:{self.db_password}{self.db_host}:{self.db_port}/{self.db_name} settings Settings()有一点容易忽略.env 文件一定要加到 .gitignore 里避免 API Key 泄露。同时 .env.example 里不要把真实密码写进去只写占位符。我见过不止一个项目因为 .env 没加进 gitignore结果密钥被推到公开仓库这个教训代价太大。2.4 日志系统初始化日志是基础设施里最容易被低估的模块。AI Agent 的调用链路很长一次完整的问数请求会经历 LLM 调用、向量检索、数据库查询等多个环节如果没有结构化的日志出了问题根本无从下手。我用 loguru 替代标准 logging配置很简单# app/utils/logger.py import sys from loguru import logger logger.remove() logger.add( sys.stdout, formatgreen{time:YYYY-MM-DD HH:mm:ss.SSS}/green | level{level: 8}/level | cyan{name}/cyan:cyan{function}/cyan:cyan{line}/cyan | level{message}/level, levelINFO, ) logger.add( logs/lcoder_{time:YYYYMMDD}.log, rotation00:00, retention30 days, levelDEBUG, )在关键的节点处打印结构化信息比如“执行节点: sql_generator输入状态: ...”同时输出 token 消耗和耗时。生产环境再接入日志采集系统把 loguru 的输出引导到 stdout由采集器统一处理。3. 模型接入层与 Prompt 基础封装3.1 LLM 统一调用封装模型接入层是 Agent 的“发动机”这里我踩过不少坑。最开始的版本是直接在各节点里调用ChatOpenAI结果发现超时、限流、格式错误这些异常散落在各处处理逻辑重复且混乱。后来抽了一个统一的LLMClient类把重试、超时、token 统计都收敛到一起。# app/llm/client.py import json import time from typing import Any from langchain_openai import ChatOpenAI from tenacity import ( retry, stop_after_attempt, wait_exponential, retry_if_exception_type, ) from app.config import settings from app.utils.logger import logger class LLMClient: def __init__(self): self._client ChatOpenAI( modelsettings.llm_model, base_urlsettings.llm_base_url or None, api_keysettings.llm_api_key, temperaturesettings.temperature, max_tokenssettings.max_tokens, timeout30, ) retry( stopstop_after_attempt(3), waitwait_exponential(multiplier1, min2, max10), retryretry_if_exception_type(TimeoutError), reraiseTrue, ) def invoke(self, messages: list[dict[str, str]]) - str: start time.time() try: response self._client.invoke(messages) usage response.usage_metadata logger.debug( fLLM 调用成功prompt_tokens{usage.get(input_tokens)}, fcompletion_tokens{usage.get(output_tokens)}, fcost{time.time() - start:.2f}s ) return response.content except TimeoutError as e: logger.warning(fLLM 调用超时准备重试: {e}) raise def bind_structured_output(self, schema: type[Any]): return self._client.with_structured_output(schema) llm_client LLMClient()这里有两个关键点。第一重试只对超时和临时性错误做模型返回内容格式错误不能盲目重试否则会浪费大量 token。第二base_url参数是为了兼容各类 OpenAI 协议接口很多国产模型都提供兼容端点留一个配置位后面切换模型会非常方便。3.2 嵌入模型与向量库接入问数项目里向量库的核心作用是做“元数据语义检索”。业务方的提问往往是口语化的比如“上个月的退货率怎么样”但数据库里的表名可能叫monthly_returns_stat直接搜是搜不到的。把字段注释、业务口径、历史经验存进向量库先做语义检索找到最相关的表和字段再交给 LLM 生成 SQL准确率会大幅提升。嵌入模型选型上我开发环境用的是 API 方式调用文本嵌入模型但要注意一点中文场景下通用英文嵌入模型的效果并不理想。如果数据字典里有大量中文描述建议换成专门的向量模型或者中文优化的嵌入模型。线上环境可以用本地部署的嵌入模型比如 bge-m3避免每次请求都走网络。向量库我用 Chroma 作为开发环境存储因为它在本地就是一个文件夹零部署成本。代码封装如下# app/rag/store.py from langchain_chroma import Chroma from langchain_openai import OpenAIEmbeddings from app.config import settings from app.utils.logger import logger embeddings OpenAIEmbeddings( modelsettings.embedding_model, base_urlsettings.llm_base_url or None, api_keysettings.llm_api_key, ) vector_store Chroma( collection_namesettings.collection_name, embedding_functionembeddings, persist_directorysettings.vector_store_path, ) def upsert_documents(docs: list[str], ids: list[str], metadatas: list[dict] | None None): vector_store.add_texts(textsdocs, idsids, metadatasmetadatas) logger.info(f写入 {len(docs)} 条文档到向量库collection{settings.collection_name}) def search_documents(query: str, k: int 5) - list[dict]: results vector_store.similarity_search_with_score(query, kk) return [ {content: doc.page_content, metadata: doc.metadata, score: score} for doc, score in results ]这里有个细节add_texts默认会重新计算所有文本的向量。如果一次性导入几百条数据字典速度还能接受但如果是增量更新建议用update_documents而不是重复 add。另外 collection 名称最好不要随便改因为在 Chroma 里切换 collection 相当于换了一个独立的向量空间之前的索引就查不到了。3.3 Prompt 模板的初始化管理Prompt 是 Agent 的灵魂但基础设施阶段只需要把模板的加载机制建好。我单独建了一个prompts.py文件把所有节点用的 Prompt 模板集中管理不散落在各节点里。加载函数用 langchain 的PromptTemplate# app/llm/prompts.py from langchain_core.prompts import PromptTemplate SQL_GENERATION_TEMPLATE PromptTemplate.from_template( 你是一名资深数据分析工程师请根据用户问题和数据表结构生成 SQL 查询语句。 用户问题: {question} 相关数据表结构: {table_schemas} 历史查询经验: {retrieved_knowledge} 要求: 1. 只生成 SQL 语句不要额外解释 2. 如果无法从提供的信息判断输出 ERROR: 信息不足 3. SQL 必须是只读查询禁止 INSERT、UPDATE、DELETE 4. 注意空值处理和数据类型转换 SQL: )集中管理的好处是后续做 Prompt 版本对比和 A/B 测试非常方便改模板不需要动业务代码。一个团队如果连 Prompt 都分散在各处后面维护成本是灾难级的。4. 数据源连接与安全管控4.1 数据库连接池配置问数 Agent 的每一次查询都需要访问业务数据库。如果为每个请求都新建连接高并发下数据库很快会被打满。SQLAlchemy 的连接池配置要提前做好# app/datasource/engine.py from sqlalchemy import create_engine from app.config import settings from app.utils.logger import logger engine create_engine( settings.db_url, pool_size10, max_overflow20, pool_pre_pingTrue, pool_recycle1800, connect_args{ connect_timeout: 5, options: -c statement_timeout10000, }, ) def get_connection(): return engine.connect()几个参数要解释一下pool_size10是连接池保持的最小连接数10 对于大多数业务库够用max_overflow20表示连接不够时可以临时新增的最大连接数防止突发流量直接打崩数据库pool_pre_pingTrue会在取出连接时先做一次连通性检查避免拿到已断开的连接pool_recycle1800强制连接 30 分钟回收一次防止数据库重启后连接失效还有一个很容易忽略的点statement_timeout。Agent 生成的 SQL 有可能是全表扫描级别的慢查询如果没有超时保护一个错误 SQL 就能把数据库拖垮。我在连接参数里直接设置了 statement_timeout10 秒从源头限制最坏情况。4.2 只读账号与 SQL 安全校验数据安全问题在问数项目里是红线。Agent 生成的 SQL 必须保证是只读的这一点不能依赖模型自觉必须在基础设施层面加两道保险。第一道保险是用数据库只读账号。给 Agent 创建的数据库账号只授予 SELECT 权限从权限层面杜绝写操作。如果公司数据库权限管控比较严可以单独建一个 Schema把需要开放的业务表做成视图Agent 只能查这些视图。第二道保险是 SQL 语法层校验。即使使用只读账号也要防止 Agent 生成“SELECT * FROM table”这种全表扫描。我写了一个简单的校验函数# app/datasource/validator.py import sqlparse from sqlparse.tokens import DML def validate_readonly_sql(sql: str) - bool: parsed sqlparse.parse(sql) for statement in parsed: for token in statement.tokens: if token.ttype in (DML,): if token.value.upper() not in (SELECT, WITH): return False return True这个函数只匹配 DML 类型的 token能拦截 INSERT、UPDATE、DELETE但对 CTE 嵌套里的写操作不一定能完全识别。更稳妥的做法是在数据库侧用 pg_rewrite 之类的机制强制改写非只读 SQL或者直接使用 PostgreSQL 的default_transaction_read_only on让事务级只读。4.3 元数据查询接口问数 Agent 需要知道数据库里“有哪些表、每个表有哪些字段、注释是什么”这些信息我统一从一个元数据接口获取。我直接用 SQLAlchemy 的 inspector 读取# app/datasource/metadata.py from sqlalchemy import inspect from app.datasource.engine import engine def get_table_schemas(table_names: list[str] | None None) - str: inspector inspect(engine) if table_names is None: table_names inspector.get_table_names() schemas [] for table in table_names: columns inspector.get_columns(table) col_desc , .join( f{col[name]} {col[type]} COMMENT {col.get(comment, )} for col in columns ) schemas.append(f表名: {table}\n字段: {col_desc}) return \n\n.join(schemas[:20])这里限制了返回表数量避免一次性把几十张表的字段全部塞给模型导致超出上下文窗口。实际项目中应该先通过向量检索召回“相关表”再只把这些表的字段结构返回给模型这样效果和效率都能兼顾。5. Agent 状态图与工具函数注册5.1 状态对象定义LangGraph 的核心是状态传递。我定义了一个AgentState所有节点读写的字段都收敛在这里# app/agent/state.py from typing import TypedDict, Annotated, Optional class AgentState(TypedDict, totalFalse): question: str session_id: str retrieved_tables: list[str] table_schemas: str sql: str sql_error: Optional[str] query_result: Optional[str] answer: str steps: list[str]这里有几个设计考量。steps字段用来记录每个节点执行的情况方便最后返回给前端展示 Agent 的“思考过程”。sql_error是专门用来做 SQL 修复的——如果执行失败节点会根据错误信息重新生成 SQL这也是 LangGraph 条件分支的典型用法。5.2 工具函数注册与实现问数 Agent 的工具函数不多但每一个都很关键。我注册了三个核心工具# app/agent/tools.py from langchain_core.tools import tool tool def search_table_schema(query: str) - str: 根据自然语言描述检索相关的数据库表结构信息 results search_documents(query, k5) return \n\n.join( f表: {item[metadata].get(table_name, unknown)}\n f字段: {item[content]} for item in results ) tool def execute_sql(sql: str) - str: 执行只读 SQL 查询并返回前 100 行结果 import pandas as pd sql_clean sql.strip().rstrip(;) try: df pd.read_sql_query( sql_clean, engine, paramsNone, ) if df.empty: return 查询结果为空 return df.head(100).to_markdown(indexFalse) except Exception as e: return fSQL 执行失败: {e}用tool装饰器之后LangChain 会自动从函数的 docstring 生成工具的描述信息。这里尤其要注意docstring 里一定要说清楚工具的用途和注意事项因为大模型判断“什么时候该用这个工具”全靠这些描述。描述写得模糊模型就会乱调用。execute_sql这里有一个生产环境的坑不能忽略把任意的 SQL 子串拼接进pd.read_sql_query虽然用的是只读账号但仍有被注入恶意查询的风险。所以线上版本一定要加上 4.2 节的校验逻辑并且对 SQL 长度做限制。5.3 LangGraph 主流程搭建主流程我搭建了一个简化版的三节点图# app/agent/graph.py from langgraph.graph import StateGraph, END from app.agent.state import AgentState from app.agent.tools import search_table_schema, execute_sql def node_schema_lookup(state: AgentState) - AgentState: if table_schemas not in state or not state[table_schemas]: schema search_table_schema.invoke(state[question]) state[steps].append(f已检索相关表结构: {schema[:100]}...) state[table_schemas] schema return state def node_sql_generation(state: AgentState) - AgentState: prompt SQL_GENERATION_TEMPLATE.format( questionstate[question], table_schemasstate[table_schemas], retrieved_knowledge, ) sql llm_client.invoke([{role: user, content: prompt}]) state[sql] sql state[steps].append(f生成 SQL: {sql}) return state def node_execute_sql(state: AgentState) - AgentState: if not validate_readonly_sql(state[sql]): state[sql_error] SQL 包含非只读操作已拒绝执行 return state result execute_sql.invoke(state[sql]) state[query_result] result state[steps].append(f执行查询: {result[:200]}) return state graph StateGraph(AgentState) graph.add_node(schema_lookup, node_schema_lookup) graph.add_node(sql_generation, node_sql_generation) graph.add_node(execute_sql, node_execute_sql) graph.set_entry_point(schema_lookup) graph.add_edge(schema_lookup, sql_generation) graph.add_edge(sql_generation, execute_sql) graph.add_edge(execute_sql, END) agent_app graph.compile()LangGraph 的图编译之后就是一个可调用的对象后续接 FastAPI 路由或者 WebSocket 都非常方便。这里展示的还是一个线性流程实际生产版我会加入“意图识别”节点和“SQL 错误修复”的循环边但基础设施阶段的骨架已经成型后续加节点就是在图上插一条边的事。6. 可观测性与常见问题排查6.1 链路追踪与调用链日志Agent 项目排错最大的难点在于链路长、状态多。用户问一句“上个月各区域销售额”背后可能是 5 次模型调用、3 次向量检索、1 次数据库查询。如果没有链路追踪出了问题根本不知道是模型生成错了 SQL还是 SQL 执行超时还是结果格式化崩了。我在项目里接了自托管的 Langfuse 来做链路追踪配置方式很简单# app/utils/trace.py from langfuse.callback import CallbackHandler from app.config import settings langfuse_handler CallbackHandler( public_keysettings.langfuse_public_key, secret_keysettings.langfuse_secret_key, hostsettings.langfuse_host, ) def get_trace_handler(): return langfuse_handler然后在调用 Agent 的时候把 handler 传进去result agent_app.invoke( {question: question, session_id: session_id, steps: []}, config{callbacks: [get_trace_handler()]}, )Langfuse 会把每次模型调用、工具调用的输入输出全部记录下来包括 token 消耗和耗时。调试的时候直接在面板上看每一步的输入输出比翻日志高效太多。如果不想自建LangSmith 也可以但它对网络环境有要求考虑数据安全我还是选了自托管方案。6.2 典型异常列表与排查方法把我在开发过程中遇到的典型问题整理了一下这些问题你在做问数智能体时大概率也会碰到现象可能原因解决方案Agent 生成 SQL 后执行报“字段不存在”元数据检索召回的表不正确模型基于错误的 schema 生成 SQL优化元数据检索结果把字段注释写得详细限制可能的表名候选集合调用模型超时网络不稳定或者模型服务负载高配置重试机制更换延迟更低的模型端点把超时时间从 30 秒改为 60 秒向量检索结果为空collection 为空或者嵌入模型调用失败检查 collection 是否存在重新执行数据字典导入确认嵌入模型 API Key 有效多轮对话串台session_id 没传或者会话状态没保存前端每次请求必须携带 session_idRedis 的会话 key 设置合理过期时间SQL 执行把数据库拖慢生成的 SQL 缺少 WHERE 条件在 Prompt 中强制要求带上时间范围条件数据库层设置 statement_timeout对慢 SQL 做拦截告警Agent 回答“我不知道”但表里其实有数据元数据检索没召回相关表增加检索召回数量优化数据字典的描述文本使用混合检索策略6.3 几个值得记住的避坑细节最后分享几个基础设施阶段容易被忽略的细节每一个都是真金白银换来的教训。第一SQL 工具返回的结果不要无限长。模型上下文窗口有限强制返回前 100 行数据并转成 Markdown够生成结论就可以了。如果业务方想看更多数据让 Agent 给出下载链接而不是把全量数据塞给模型。第二会话状态和记忆是两回事。会话状态是指 session 级别的临时数据放到 Redis 里设一个合理的过期时间比如 30 分钟长期偏好类记忆需要单独存储。不要把两者混在一个 Redis key 里否则会越存越大。第三所有外部依赖都要设置超时。模型调用有超时数据库连接有超时Redis 操作也要有超时。一个环节没有超时控制整个请求就可能卡死在那里。我在 Redis 客户端初始化时也统一配置了 socket_connect_timeout 和 socket_timeout。第四本地环境和生产环境的向量库不要共用。开发环境用的 Chroma 本地文件里全是测试数据如果直接部署到生产会污染线上的检索结果。建议在配置里用不同的 collection 名称本地叫lcoder_dev生产叫lcoder_prod。第五日志一定要记录 token 消耗。问数类 Agent 的 token 成本比普通对话机器人高很多原因是一次问题要触发多次模型调用。我在日志里记录了每一步的 token 数按天汇总对控制成本非常有帮助。如果上线后发现成本超标优先检查是不是 SQL 修复循环走太多轮了可以通过限制循环次数来控成本。基础设施搭完之后后面加业务逻辑就像搭积木。我在实际开发过程中最大的体会是前期多花点时间把这些基础模块打磨好后面调试 Agent 策略时能省下几倍的精力。特别是链路追踪和日志如果没有它们调一个 SQL 生成问题可能要抓瞎一整天有了它们十分钟就能定位到问题出在哪个节点。下一步我会继续写工具层的增强和 LangGraph 条件分支的完整实现包括 SQL 自修复机制和复杂查询的多轮改写策略这个部分完成后LCODER 就能应对更真实、更刁钻的业务问题了。