这个系列终于走到第二篇了。上一篇我们把问数项目智能体的整体规划讲清楚了包括要解决什么问题、整体流程长什么样、技术栈往哪个方向选。当时就有不少朋友留言说规划看着明白但真正自己动手搭环境的时候各种版本冲突、依赖报错、模型接不上的问题一大堆半天搭不下来。这一篇我们就来解决这个问题专注把基础设施这一层彻底搞定。所谓基础设施不是说装个Python、装个MySQL那么简单。对于问数项目这种典型的AI Agent应用基础设置至少包含三块数据层让Agent能查询和理解你的业务库、模型层把大模型API稳定地接进来、Agent运行时层支撑Agent状态流转、工具调用、记忆存储的代码框架。这三块搭稳了后面做Agent逻辑、做提示词优化、做工具扩展才有底气否则后面每进一步都会返工。适合什么人看如果你打算从零开始搭一个能查数据库、能回答业务问题的AI Agent或者你看了一堆Agent理论但不知道怎么落地这篇就是个能照着敲的实操手册。我不会只丢一堆命令会把每个选择背后的理由、版本怎么定、踩了哪些坑都讲清楚。1. 先想清楚再动手问数智能体的整体设计与技术选型1.1 问数项目到底要解决什么问题问数项目这个名字听起来挺高大上本质就一句话让用户用自然语言提问系统自动去数据库里把答案查出来再以人能看懂的方式返回。典型例子是“上个月华北区的销售额是多少环比增长了多少”过去这需要数据分析师写SQL、跑数、做图表现在交给Agent去做。但这里有个关键点容易搞混问数智能体不是简单套一个Text2SQL的开源模型就完事了。Text2SQL解决的是“自然语言变成SQL”这一步可实际场景远不止这一步。用户的问题可能是模糊的比如“最近销售情况怎么样”你得先搞清楚“最近”是多久、看什么指标、按什么维度拆。用户可能问一个表里根本没有的字段你得知道怎么委婉地告诉他查不了。SQL跑出来结果异常你得能判断是数据问题还是SQL写错了。这些都需要Agent具有“多步推理 工具调用 结果校验”的能力而不是一条直线走到底。所以我的设计思路是把问数过程拆成几个标准环节——意图理解、查元数据有哪些表和字段、生成SQL、执行SQL、解析结果、组织回答。每个环节可以由大模型驱动也可以由代码控制Agent负责在中间做决策和调度。这个架构在选型时给了我一个很重要的判断标准我需要的是一个能灵活编排状态的框架而不是一套写死的流程代码。1.2 技术方案选型为什么是LangGraph而不是硬编码流程说到Agent开发框架现在市面上可选的东西很多。LangChain、LangGraph、Spring AI、Coze这类低代码平台、还有各种国产框架。对于问数项目这种偏工程化、需要深度定制的场景我选了LangGraph。原因很简单。问数流程看着固定实际跑起来会有大量分支。比如第一步意图识别如果识别到用户只是在闲聊而不是问数那就该走闲聊分支如果识别到需要先看表结构才能写SQL就得先调用元数据查询工具SQL执行报错还得自动修复重试。这种复杂的状态流转用硬编码if-else写两三个分支还好写五六个分支代码就烂成一团了每次加个新功能都要动原来的流程。LangGraph的核心抽象是状态图。你把整个问数流程画成一张图节点是逻辑处理单元比如“生成SQL”是一个节点“执行SQL”是一个节点边是流转条件节点之间通过一个共享的State对象传递数据。这样做的好处有两个一是流程逻辑可视化图和代码一一对应调试的时候脑子里有一张地图二是每个节点相对独立我改“生成SQL”这个节点的内部实现不会影响其他节点。对于问数项目这种后续要不断迭代工具、加新能力的场景这种可扩展性太关键了。有人会问为什么不用LangChain直接做LangChain的Chain更适合线性流程Agent部分虽然也能跑但它是隐式循环内部帮你实现了一个AgentExecutor调试和自定义都比较别扭。LangGraph把循环、分支、记忆这些机制显式暴露出来更适合我们这种对流程控制有要求的场景。至于Coze这类低代码平台快速搭个Demo确实方便但要接内部数据库、做复杂的权限管理和深度定制平台限制就会冒出来实战项目我还是倾向代码方案。2. 基础设施全景数据层、模型层、Agent运行时三层打底2.1 数据层让Agent能“看懂”你的库数据层是问数项目的地基。Agent要回答业务问题前提是它能访问到一个结构清晰、元数据完备的数据库。很多问数项目死在全球第一个环节——Agent不知道你的表里有哪些字段、字段是什么意思生成的SQL自然就是在瞎猜。先说数据库选型。问数项目最常见的是接MySQL或PostgreSQL中小团队的业务数据基本都在这两个库里。我这边示例用MySQL 8.0一方面是存量数据在这里另一方面MySQL的生态工具最成熟排查问题容易。如果你用的是PostgreSQL下面讲的思路完全适用只是连接驱动和方言略有差异。要让Agent“看懂”数据库核心工作是做元数据管理。所谓元数据就是关于数据的数据——表名、字段名、字段类型、注释、枚举值的含义、指标的计算口径。举个例子订单表里有一个字段叫status存的值是0、1、2如果不在元数据里说明0代表待支付、1代表已支付、2代表已取消大模型写SQL的时候就会瞎猜要么猜成字符串要么猜不清楚取值范围。所以我在建表的时候就要求所有表和字段必须有COMMENT注释除此之外还会维护一份指标字典把常用的业务口径写清楚比如“销售额订单金额-退款金额”。实际搭建的时候我建议你建立一个单独的元数据表来存这些信息也可以在Agent的工具层考虑我放在后面工具设计部分详细讲。这里先强调一点有些读者测试的时候用的是随便建的demo表字段叫a、b、c没有注释然后抱怨Agent生成SQL不准。这真的不是Agent的问题是你没给它提供足够的信息。让Agent看懂数据和数据本身的质量同等重要。2.2 模型层统一接口灵活切换模型层负责把大模型能力接进来。问数项目里有两个地方需要大模型主对话链路理解意图、生成SQL、组织回答和可选的信息抽取比如从非结构化输入里抽查询条件。如果你后续打算做向量检索比如根据历史问答推荐相似问题那还要接Embedding模型。接大模型这件事我强烈建议代码里只写一套OpenAI兼容接口然后通过环境变量的方式切换具体模型。原因很简单现在国内外主流模型厂商基本都提供OpenAI兼容的HTTP接口你把base_url和api_key换成对应厂商的值就能切过去。这样做的好处是你可以在开发环境用某种模型调试在和生产环境模型不一致时快速切换或者在模型厂商出问题的时候应急换另一家的模型代码一行都不用动。具体到模型选择问数项目的核心是SQL生成和工具调用所以我优先选工具调用Function Calling能力强的模型这一类能力直接决定了Agent能否稳定地使用工具。另外上下文窗口要够大因为要往里塞表结构信息和多轮对话历史。在实操部分我会给出一个最小的模型客户端封装支持从配置里同时读取base_url、api_key、model_name三个参数。2.3 Agent运行时LangGraph环境搭建Agent运行时这一层说白了就是支撑Agent跑起来的代码环境。我用LangGraph作为核心框架所以这块的主要工作是建Python虚拟环境、装依赖、把LangGraph的状态机制用一个最小例子跑通。Python版本建议直接用3.11或3.12。我踩过Python 3.8跑新版本LangChain相关依赖时各种报错的坑提醒一句项目新建直接上3.11省心。这里不是越新越好3.13刚出来的时候不少依赖还没跟上稳一点选3.11或3.12。依赖安装是整个搭建过程里最容易让人崩溃的一步。LangGraph这个生态迭代速度极快包名特别多而且pydantic版本、langchain-core版本经常互相打架。我自己总结了一个原则不要图省事一下装一堆包先按照“跑通最小例子”需要的量来装跑通了再逐步加。核心依赖大概是这几类langgraph本体、langchain-openai提供ChatOpenAI模型封装、openai SDK、pydantic和pydantic-settings配置管理、SQLAlchemy和PyMySQL数据库操作、python-dotenv环境变量加载。具体版本我在实操部分会给一份可以直接用的requirements文件。LangGraph还有一个特殊机制叫Checkpoint检查点它能把Agent每一步的状态保存下来这样支持多轮对话的记忆恢复、人工中断审核、异常恢复重试等功能。基础搭建阶段我们可以先把它的存储用一个本地SQLite或PostgreSQL表搞定不用Redis那么重。后面要做高并发、多实例部署再考虑把存储换成Redis或其他共享存储。3. 动手实操从零把基础设施跑起来3.1 环境初始化与项目目录规范我先定义一个清晰的项目目录。问数项目叫question_agent目录结构如下question_agent/ ├── .env # 环境变量不进仓库 ├── .env.example # 环境变量模板进仓库 ├── .gitignore ├── requirements.txt ├── README.md ├── agent/ # Agent相关代码 │ ├── __init__.py │ ├── state.py # LangGraph状态定义 │ ├── llm.py # 模型客户端封装 │ ├── tools/ # 工具集合 │ │ ├── __init__.py │ │ ├── db.py # 数据库工具 │ │ └── metadata.py # 元数据查询工具 │ └── graph.py # Agent图定义 ├── config/ │ ├── __init__.py │ └── settings.py # 配置类 ├── scripts/ │ ├── init_db.sql # 建表SQL │ └── test_connection.py # 连接测试脚本这个目录不是拍脑袋定的。把代码按agent、config、scripts分层是为了让“模型封装”“工具定义”“流程编排”各归其位后面扩展的时候不用在一个几百行的文件里到处找。很多初学者喜欢把代码全部堆在一个main.py里跑通Demo没问题但项目一复杂就完蛋。接下来是虚拟环境和核心依赖安装cd question_agent python3.11 -m venv venv source venv/bin/activate pip install --upgrade pip pip install langgraph0.2.0 \ langchain-openai0.2.0 \ openai1.40.0 \ pydantic2.7.0 \ pydantic-settings2.3.0 \ python-dotenv1.0.0 \ sqlalchemy2.0.30 \ pymysql1.1.0 \ pandas2.2.0注意版本号我标的是经过验证的下限版本。LangGraph 0.2版本之后API变化较大如果你是照着老教程用StateGraph的旧写法很可能报错。建议直接以新版为准下面我的示例代码就是新版写法。这里还有一个值得说的点为什么要用 virtualenv 而不是 conda 或直接用系统Python。conda也能用但它的包管理有时候会和pip产生冲突尤其LangChain生态有些依赖版本比较敏感。virtualenv加pip的组合最干净出了问题重建环境也快。一个项目一个虚拟环境这是基础设施阶段就要养成的习惯。3.2 配置管理与数据库初始化配置管理这块核心原则就一条代码不掺密钥配置不进仓库。我用.env文件存所有环境变量然后用 pydantic-settings 在代码里统一读取和校验。.env.example文件内容如下你复制一份改成.env填上真实值# 模型配置 LLM_PROVIDERopenai LLM_BASE_URLhttps://api.example.com/v1 LLM_API_KEYsk-your-key-here LLM_MODELgpt-4o-mini LLM_TEMPERATURE0 # 数据库配置 DB_HOST127.0.0.1 DB_PORT3306 DB_USERagent_reader DB_PASSWORDyour-password DB_NAMEquestion_agent_demo DB_ECHOfalse # Agent配置 AGENT_MAX_ITERATIONS10 AGENT_VERBOSEtrue对应配置类写在config/settings.py里from pydantic_settings import BaseSettings, SettingsConfigDict class Settings(BaseSettings): model_config SettingsConfigDict(env_file.env, env_file_encodingutf-8) llm_provider: str openai llm_base_url: str https://api.example.com/v1 llm_api_key: str llm_model: str gpt-4o-mini llm_temperature: float 0.0 db_host: str 127.0.0.1 db_port: int 3306 db_user: str root db_password: str db_name: str question_agent_demo db_echo: bool False agent_max_iterations: int 10 agent_verbose: bool True settings Settings()这里我特别解释一下LLM_TEMPERATURE0这个配置。问数项目里我们要生成SQL、做意图判断这些都是偏确定性任务温度设成0可以让大模型的输出尽量稳定降低随机性带来的SQL差异。我知道有些同学喜欢把温度调高觉得“更有创造性”但在Agent工具调用场景里这是万万要不得的你绝不希望同一个问题两次生成不一样的SQL。数据库初始化方面我先给一个小型的电商销售demo库。为什么用销售数据做demo因为销售这个场景大家都很熟悉订单、用户、产品、区域这些概念不用过多解释方便验证Agent的问答效果。scripts/init_db.sql核心建表语句CREATE DATABASE IF NOT EXISTS question_agent_demo DEFAULT CHARACTER SET utf8mb4; USE question_agent_demo; CREATE TABLE dim_product ( product_id INT PRIMARY KEY AUTO_INCREMENT COMMENT 商品ID, product_name VARCHAR(128) NOT NULL COMMENT 商品名称, category VARCHAR(64) NOT NULL COMMENT 商品类目, unit_price DECIMAL(10,2) NOT NULL COMMENT 单价元, status TINYINT NOT NULL DEFAULT 1 COMMENT 状态1上架0下架 ) COMMENT 商品维度表; CREATE TABLE dim_region ( region_id INT PRIMARY KEY AUTO_INCREMENT COMMENT 区域ID, region_name VARCHAR(64) NOT NULL COMMENT 区域名称, province VARCHAR(64) NOT NULL COMMENT 省份 ) COMMENT 区域维度表; CREATE TABLE fact_order ( order_id BIGINT PRIMARY KEY AUTO_INCREMENT COMMENT 订单ID, order_no VARCHAR(32) NOT NULL COMMENT 订单编号, order_date DATETIME NOT NULL COMMENT 下单时间, product_id INT NOT NULL COMMENT 商品ID, region_id INT NOT NULL COMMENT 区域ID, quantity INT NOT NULL COMMENT 销售数量, amount DECIMAL(12,2) NOT NULL COMMENT 成交金额元, status TINYINT NOT NULL DEFAULT 1 COMMENT 订单状态0已取消1已完成2售后中 ) COMMENT 订单事实表;看着是不是很简单但这里有几个细节值得注意。第一所有表、字段都有中文COMMENT这直接决定了大模型能不能理解表结构是我反复强调的元数据建设基础。第二字段命名用下划线风格虽然拼音乱写也能用但规范命名能减少模型误解。第三金额字段用DECIMAL而不是FLOAT避免浮点精度问题——这种问题一旦发生Agent算出的销售额差了0.1元你排查起来会非常崩溃。建完表之后我还建议写入一部分示例数据。有了数据Agent跑SQL才有结果可返回你才能看到完整的链路效果。示例数据不用多覆盖几个类目、几个区域、近几个月的订单就行具体插入语句这里不展开了你按自己的理解造一批合理的假数据即可。造数据的时候注意日期字段要覆盖最近至少3个月订单状态不要全是已完成要有一些已取消和售后中的这样Agent才能遇到“需要排除已取消订单”这类真实问题。3.3 模型客户端封装模型封装的目标是在你代码的任何一个地方一行代码拿到一个可调用的LLM对象而且切换模型不用改业务代码。agent/llm.pyfrom langchain_openai import ChatOpenAI from config.settings import settings def get_llm(): return ChatOpenAI( modelsettings.llm_model, api_keysettings.llm_api_key, base_urlsettings.llm_base_url, temperaturesettings.llm_temperature, timeout60, max_retries2, )你可能注意到这里用langchain_openai的ChatOpenAI而不是直接调用openai SDK。原因很简单LangGraph生态里的模型节点、工具调用绑定都默认使用LangChain的模型封装格式。直接用OpenAI SDK当然也能调但还得自己处理消息格式转换、工具调用协议解析纯属给自己找麻烦。用ChatOpenAI作为统一入口后续所有LangGraph特性都天然支持。不过千万别把这些写死在代码里所有参数必须走配置。实际开发中经常遇到测试环境用一个模型、生产环境用另一个模型、模型供应商临时出故障要紧急切换的情况配置化之后只需要改.env里的变量然后重启服务。写一个测试脚本验证模型能不能通python -c from agent.llm import get_llm llm get_llm() resp llm.invoke(用一句话介绍你自己) print(resp.content) 如果能正常返回一句话说明模型链路通了。这一步挂了就先解决它不要着急往下走。3.4 数据库连接工具与基础验证有了配置和模型接着写数据库访问层。这里用SQLAlchemy 2.0的engine加连接池应对Agent生成的并发查询足够了。同时考虑到安全建议给Agent创建一个只读数据库账号这非常关键。建一个查询代理账号并授权CREATE USER agent_reader% IDENTIFIED BY your-password; GRANT SELECT ON question_agent_demo.* TO agent_reader%; FLUSH PRIVILEGES;为什么一定要只读账号因为Agent生成SQL是有一定随机性的如果不小心让它拿到写权限一句DELETE FROM fact_order就能把几年的业务数据全干没了。这里不是假设Agent会故意搞破坏而是大模型无法保证100%正确我们必须在权限层面把所有危险操作挡在门外。这个习惯要在一开始就养成。agent/tools/db.py提供一个最简单的SQL执行工具from sqlalchemy import create_engine, text from config.settings import settings engine create_engine( fmysqlpymysql://{settings.db_user}:{settings.db_password}{settings.db_host}:{settings.db_port}/{settings.db_name}?charsetutf8mb4, pool_size5, pool_recycle3600, echosettings.db_echo, ) def run_sql(sql: str) - str: 执行只读SQL返回JSON格式字符串结果。 with engine.connect() as conn: result conn.execute(text(sql)) columns list(result.keys()) rows [dict(zip(columns, r)) for r in result.fetchmany(200)] return json.dumps(rows, ensure_asciiFalse, defaultstr)这里有几个细节大家可能不注意。fetchmany(200)是限制最多取200行防止Agent写了一个没有WHERE条件的全表查询把内存打爆。返回值用JSON字符串是因为LangGraph的State传输要求序列化友好后面接工具返回给大模型也方便。defaultstr是为了处理datetime和Decimal这些不能直接JSON序列化的类型这个坑我印象特别深第一次跑通之前在这里卡了一个多小时。写连接测试脚本# scripts/test_connection.py from agent.tools.db import run_sql if __name__ __main__: result run_sql(SELECT COUNT(*) AS cnt FROM fact_order) print(result)能输出订单总数数据链路就通了。3.5 第一个最小Agent跑通“问题→SQL→结果”链路基础设施全部就位后最后一步是跑一个最小可用的Agent验证整条链路。这一步的核心目标是让LangGraph按流程走一遍“理解问题→生成SQL→执行SQL→总结答案”我们不做复杂的工具调度先把链路打通。LangGraph的新版API实现如下from typing import TypedDict, Literal from langgraph.graph import StateGraph, END from langchain_core.messages import HumanMessage, SystemMessage from agent.llm import get_llm from agent.tools.db import run_sql SYSTEM_PROMPT 你是一个数据分析助手。根据用户的问题生成SQL查询语句。 要求 1. 只查询表 fact_order, dim_product, dim_region 2. 只生成SELECT语句禁止任何修改操作 3. 表字段有中文注释注意理解字段含义 4. 只输出SQL不要解释 class State(TypedDict): question: str sql: str result: str answer: str def generate_sql(state: State) - dict: llm get_llm() resp llm.invoke([ SystemMessage(contentSYSTEM_PROMPT), HumanMessage(contentstate[question]), ]) return {sql: resp.content.strip()} def execute_sql(state: State) - dict: result run_sql(state[sql]) return {result: result} def generate_answer(state: State) - dict: llm get_llm() resp llm.invoke([ SystemMessage(content根据用户问题和查询结果用通俗中文回答用户的问题。如果查询结果为空明确告知查不到数据。), HumanMessage(contentf问题{state[question]}\n查询结果{state[result]}), ]) return {answer: resp.content} app StateGraph(State) app.add_node(generate_sql, generate_sql) app.add_node(execute_sql, execute_sql) app.add_node(generate_answer, generate_answer) app.set_entry_point(generate_sql) app.add_edge(generate_sql, execute_sql) app.add_edge(execute_sql, generate_answer) app.add_edge(generate_answer, END) agent app.compile() if __name__ __main__: result agent.invoke({question: 上个月假设8月卖出多少件商品}) print(SQL:, result[sql]) print(结果:, result[result]) print(回答:, result[answer])这个最小Agent看起来简单但它验证了三件重要的事情模型能按提示词生成SQL、能连上数据库执行查询、能根据结果生成自然语言回答。很多初学者一上来就想做带工具调用的复杂Agent结果链路某一段是坏的根本分不清是哪儿出了问题。先跑通这个最小环相当于给整个系统做了一个冒烟测试后面再往上加工具、加分支、加记忆都是在稳定的地基上盖楼。跑通之后你可以在app.compile()时传入checkpointer参数来启用对话记忆LangGraph会保存每次运行的State这样下一个问题来的时候还能参考上一个SQL的执行经验。基础阶段先用MemorySaver()内存模式就够了后面需要持久化再存到SQLite或Redis。4. 工程化细节日志、异常、安全一个都不能少4.1 日志与可观测性Agent每一步都要留痕Agent开发和传统后端开发有一个非常大的不同传统接口出问题了看报错栈基本能定位Agent出问题往往不是“报错”而是“答错了”或者“绕远了”。比如一次查询结果不对你根本不知道是哪一层出了问题——是意图理解偏了SQL生成错了还是结果解析错了所以日志和可观测性是Agent项目基础设施的重要组成。我建议至少做三件事。第一每个节点入口和出口打日志记录节点名、输入的关键字段、输出的关键字段耗时也顺手记下来。第二打开LangGraph的debug模式它会打印完整的节点调用序列和状态流转。第三生产环境建议接入LangSmith或者自建一套提示词调用日志把每次大模型的输入输出都存下来方便事后复盘。当前基础设施阶段先做到前两点。用logging模块来做配置在入口文件里统一设置import logging logging.basicConfig( levellogging.INFO, format%(asctime)s | %(levelname)s | %(name)s | %(message)s, ) logger logging.getLogger(agent) # 节点里这样用 logger.info(generate_sql start, question%s, state[question])我见过不少项目日志只在出问题时才打印一行except Exception as e: print(e)这对Agent项目是远远不够的。务必让每一步都有迹可循这是排查一切疑难杂症的前提。4.2 异常处理与超时控制高可用不是上线后的事基础设施阶段的另一个工程化任务是异常处理。问数项目最常见的异常有这几类数据库连不上、SQL语法错误、查询超时、模型API限流、Token超限。每一类都应该有对应的处理策略。以SQL执行环节为例我们要区分两类异常。一类是SQL语法错误这种往往是大模型拼错了表名、字段名或语法结构可以在异常里捕获后带着错误信息让模型重新生成一次。一类是执行超时比如大模型写了一个笛卡尔积查询这种重试没意义应该直接返回友好错误并终止。代码层面给run_sql加上超时控制def run_sql(sql: str, max_rows: int 200, timeout: int 10) - str: try: with engine.connect() as conn: conn.execute(text(SET SESSION MAX_EXECUTION_TIME10000)) result conn.execute(text(sql)) ... except Exception as e: return json.dumps({error: str(e)}, ensure_asciiFalse)SET SESSION MAX_EXECUTION_TIME是MySQL端超时如果SQL跑超过10秒就直接终止不浪费数据库资源。对于更严格的场景还可以在数据库账号层面加一条MySQL的max_statement_time限制双重保险。大模型API调用也需要超时和重试。我在get_llm()里已经设了timeout60, max_retries2这个配置基本够用。再往上可以加指数退避的重试策略但基础阶段不必弄得太复杂能保证“失败能感知、不会无限卡死”就够了。4.3 配置安全与密钥管理不要在GitHub上裸奔这一节虽然短但我觉得还是值得单独说一下因为在实际项目里这个问题太常见了。密钥泄漏通常是安全意识问题而不是技术问题。第一.env文件永远不要提交到Git仓库。创建项目那一瞬间就写一个靠谱的.gitignore.env venv/ __pycache__/ *.pyc .DS_Store第二仓库里只放.env.example里面用占位符代替真实密钥。这样新同事拉完代码复制一份.env.example为.env再填自己的配置一分钟就能跑起来而不会把你本地的密钥也一起拉走。第三密钥要有环境隔离。开发、测试、生产环境用不同的.env文件比如.env.dev和.env.prod启动时通过环境变量指定加载哪个。这能防止你在测试环境误操作指向生产数据库。第四如果真的泄漏了别心存侥幸马上到密钥平台吊销并换新没有第二条路。5. 常见问题与排查技巧搭建期最容易踩的坑基础设施搭建阶段的问题大同小异我把问数项目搭建时大家问得最多的几类问题整理成一个速查表都是我自己或朋友实际踩过的。现象可能原因排查方法与解决方案调用模型API报401/403密钥错误或没写入.env文件检查.env是否已加载打印 settings.llm_api_key 确认值确认密钥有无过期模型调用超时base_url填错、网络不通、模型名不存在先用curl直接测API接口确认模型名和厂商平台一致检查timeout配置pydantic版本冲突langchain导入报错依赖版本互相不兼容按requirements.txt版本安装不要无脑装最新版重建虚拟环境连接MySQL报Cant connectMySQL没启动、端口不对、账号权限不对先mysql命令行原生连一次确认排除MySQL自身问题查user表确认账号存在检查bind-addressAgent生成的SQL语法对但语义错元数据不清晰检查表字段COMMENT是否完整确认提示词里给了足够字段说明在提示词中列出常用的字段枚举值SQL执行返回乱码连接串charset没设置utf8mb4连接engine的URL加?charsetutf8mb4确认数据库表本身是utf8mb4LangGraph节点重复执行或死循环State传递异常或边定义错误打开debug模式打印节点顺序检查是否忘记设置END边检查条件边返回值类型返回结果字段缺失SQL执行fetchmany后丢了列类型确认result.keys()能取到列名用defaultstr处理Decimal和datetime展开讲几个我在实践中印象最深的坑。第一个是 pydantic 版本冲突。LangChain生态早期对pydantic v1和v2的兼容性很混乱如果你在一个已经有pydantic v1依赖的旧环境里强行装LangGraph会报各种莫名其妙的属性错误。我自己的做法是项目初始化直接建全新虚拟环境pydantic统一装v2版本依赖用requirements锁定不随意升级。第二个是模型名和API不匹配。很多人用第三方模型服务以为自己填的模型名一定能用。实际上不同的服务商模型名差异很大有的叫gpt-4o有的叫qwen-plus有的叫deepseek-chat填错了虽然HTTP能通但会提示model not found。排查办法就是先用curl或Postman调一次接口别一上来就怪代码。第三个是Agent生成的SQL带着多余内容。大模型有时候会“好心”在SQL前面加一句“好的以下是你要的SQL”或者在结尾加分号和注释。这在提示词里强调“只输出SQL不要解释”能大大缓解但仍有漏网之鱼。更稳妥的办法是在执行前对SQL做一个清理去掉Markdown代码块标记去掉行首行尾的杂质import re def clean_sql(sql: str) - str: sql sql.strip() sql re.sub(r^sql|^|$, , sql).strip() sql re.sub(r;$, , sql) return sql这三行代码看着简单但能把你从大模型的“好心”里救出来。6. 从基础设施到真正可用下一步往哪儿扩基础设施跑通之后问数项目还远不算完。我个人习惯是基础链路一旦通了马上开始迭代“工具层”因为Agent后续的所有能力几乎都建立在工具之上。第一个要加的工具是元数据查询工具。最小值版本可以做一个get_table_schema(table_name)工具让Agent在生成SQL之前先去查一下相关的表结构。这比把所有表结构硬塞到提示词里好得多——一方面省Token另一方面当表特别多、字段特别多的时候只靠系统提示词塞信息模型根本记不住那么多。你可以在第3节最小Agent的基础上把generate_sql节点改造为“先调用元数据工具再生成SQL”。第二个要加的是多轮澄清机制。用户问题模糊的时候比如“最近销售怎么样”Agent应该主动反问“你是指销售额还是销量最近是指近7天还是近30天”这个能力需要Agent具有多轮对话的记忆能力也就是在第3节提到的checkpointer机制的基础上实现。第三个值得关注的是MCP协议。MCPModel Context Protocol如果你还没接触过可以把它理解成AI应用里的“USB接口标准”——你按照这个协议封装数据源和工具任何支持MCP的Agent客户端都能直接即插即用。问数项目里的数据库查询工具、元数据工具都可以考虑封装成MCP服务这样未来如果要把同一个能力复用到其他Agent项目里就不用重写一遍。这个方向我已经在探索了后面专门写一篇。还有一点是很现实的如果问数项目要接给业务团队用你肯定不希望业务同学每次问完都等十几秒。系统延迟优化要靠缓存和异步把高频问题对应的查询结果缓存到Redis遇到相同或相似的问题直接命中缓存。这一步属于性能基础设施的范畴等并发量真的起来了再做也不迟。说到最后我再分享一点自己的真实体会。基础设施搭建这件事看起来没有写Agent逻辑那么“高级”但它的好坏直接决定了你后期开发的幸福感。我见过太多项目Demo阶段跑得飞起一到接真实数据库、接真实业务场景就各种崩根本原因就是基础没打牢——配置写死、没有日志、权限混乱、依赖一堆坑。反过来如果你愿意在这层多花点心思把配置、目录、日志、安全、异常处理一次做到位后续所有Agent能力的迭代都会非常顺畅。搭建过程里如果遇到问题优先怀疑版本和配置这两个是最大的坑源。把这一篇的内容都跑通了咱们下一篇就可以正式开搞Agent的核心逻辑——工具调用、状态编排、多轮对话。一个真正能用的问数项目已经离你不远了。
AI Agent问数项目基础设施搭建：从零跑通数据、模型与LangGraph运行时
这个系列终于走到第二篇了。上一篇我们把问数项目智能体的整体规划讲清楚了包括要解决什么问题、整体流程长什么样、技术栈往哪个方向选。当时就有不少朋友留言说规划看着明白但真正自己动手搭环境的时候各种版本冲突、依赖报错、模型接不上的问题一大堆半天搭不下来。这一篇我们就来解决这个问题专注把基础设施这一层彻底搞定。所谓基础设施不是说装个Python、装个MySQL那么简单。对于问数项目这种典型的AI Agent应用基础设置至少包含三块数据层让Agent能查询和理解你的业务库、模型层把大模型API稳定地接进来、Agent运行时层支撑Agent状态流转、工具调用、记忆存储的代码框架。这三块搭稳了后面做Agent逻辑、做提示词优化、做工具扩展才有底气否则后面每进一步都会返工。适合什么人看如果你打算从零开始搭一个能查数据库、能回答业务问题的AI Agent或者你看了一堆Agent理论但不知道怎么落地这篇就是个能照着敲的实操手册。我不会只丢一堆命令会把每个选择背后的理由、版本怎么定、踩了哪些坑都讲清楚。1. 先想清楚再动手问数智能体的整体设计与技术选型1.1 问数项目到底要解决什么问题问数项目这个名字听起来挺高大上本质就一句话让用户用自然语言提问系统自动去数据库里把答案查出来再以人能看懂的方式返回。典型例子是“上个月华北区的销售额是多少环比增长了多少”过去这需要数据分析师写SQL、跑数、做图表现在交给Agent去做。但这里有个关键点容易搞混问数智能体不是简单套一个Text2SQL的开源模型就完事了。Text2SQL解决的是“自然语言变成SQL”这一步可实际场景远不止这一步。用户的问题可能是模糊的比如“最近销售情况怎么样”你得先搞清楚“最近”是多久、看什么指标、按什么维度拆。用户可能问一个表里根本没有的字段你得知道怎么委婉地告诉他查不了。SQL跑出来结果异常你得能判断是数据问题还是SQL写错了。这些都需要Agent具有“多步推理 工具调用 结果校验”的能力而不是一条直线走到底。所以我的设计思路是把问数过程拆成几个标准环节——意图理解、查元数据有哪些表和字段、生成SQL、执行SQL、解析结果、组织回答。每个环节可以由大模型驱动也可以由代码控制Agent负责在中间做决策和调度。这个架构在选型时给了我一个很重要的判断标准我需要的是一个能灵活编排状态的框架而不是一套写死的流程代码。1.2 技术方案选型为什么是LangGraph而不是硬编码流程说到Agent开发框架现在市面上可选的东西很多。LangChain、LangGraph、Spring AI、Coze这类低代码平台、还有各种国产框架。对于问数项目这种偏工程化、需要深度定制的场景我选了LangGraph。原因很简单。问数流程看着固定实际跑起来会有大量分支。比如第一步意图识别如果识别到用户只是在闲聊而不是问数那就该走闲聊分支如果识别到需要先看表结构才能写SQL就得先调用元数据查询工具SQL执行报错还得自动修复重试。这种复杂的状态流转用硬编码if-else写两三个分支还好写五六个分支代码就烂成一团了每次加个新功能都要动原来的流程。LangGraph的核心抽象是状态图。你把整个问数流程画成一张图节点是逻辑处理单元比如“生成SQL”是一个节点“执行SQL”是一个节点边是流转条件节点之间通过一个共享的State对象传递数据。这样做的好处有两个一是流程逻辑可视化图和代码一一对应调试的时候脑子里有一张地图二是每个节点相对独立我改“生成SQL”这个节点的内部实现不会影响其他节点。对于问数项目这种后续要不断迭代工具、加新能力的场景这种可扩展性太关键了。有人会问为什么不用LangChain直接做LangChain的Chain更适合线性流程Agent部分虽然也能跑但它是隐式循环内部帮你实现了一个AgentExecutor调试和自定义都比较别扭。LangGraph把循环、分支、记忆这些机制显式暴露出来更适合我们这种对流程控制有要求的场景。至于Coze这类低代码平台快速搭个Demo确实方便但要接内部数据库、做复杂的权限管理和深度定制平台限制就会冒出来实战项目我还是倾向代码方案。2. 基础设施全景数据层、模型层、Agent运行时三层打底2.1 数据层让Agent能“看懂”你的库数据层是问数项目的地基。Agent要回答业务问题前提是它能访问到一个结构清晰、元数据完备的数据库。很多问数项目死在全球第一个环节——Agent不知道你的表里有哪些字段、字段是什么意思生成的SQL自然就是在瞎猜。先说数据库选型。问数项目最常见的是接MySQL或PostgreSQL中小团队的业务数据基本都在这两个库里。我这边示例用MySQL 8.0一方面是存量数据在这里另一方面MySQL的生态工具最成熟排查问题容易。如果你用的是PostgreSQL下面讲的思路完全适用只是连接驱动和方言略有差异。要让Agent“看懂”数据库核心工作是做元数据管理。所谓元数据就是关于数据的数据——表名、字段名、字段类型、注释、枚举值的含义、指标的计算口径。举个例子订单表里有一个字段叫status存的值是0、1、2如果不在元数据里说明0代表待支付、1代表已支付、2代表已取消大模型写SQL的时候就会瞎猜要么猜成字符串要么猜不清楚取值范围。所以我在建表的时候就要求所有表和字段必须有COMMENT注释除此之外还会维护一份指标字典把常用的业务口径写清楚比如“销售额订单金额-退款金额”。实际搭建的时候我建议你建立一个单独的元数据表来存这些信息也可以在Agent的工具层考虑我放在后面工具设计部分详细讲。这里先强调一点有些读者测试的时候用的是随便建的demo表字段叫a、b、c没有注释然后抱怨Agent生成SQL不准。这真的不是Agent的问题是你没给它提供足够的信息。让Agent看懂数据和数据本身的质量同等重要。2.2 模型层统一接口灵活切换模型层负责把大模型能力接进来。问数项目里有两个地方需要大模型主对话链路理解意图、生成SQL、组织回答和可选的信息抽取比如从非结构化输入里抽查询条件。如果你后续打算做向量检索比如根据历史问答推荐相似问题那还要接Embedding模型。接大模型这件事我强烈建议代码里只写一套OpenAI兼容接口然后通过环境变量的方式切换具体模型。原因很简单现在国内外主流模型厂商基本都提供OpenAI兼容的HTTP接口你把base_url和api_key换成对应厂商的值就能切过去。这样做的好处是你可以在开发环境用某种模型调试在和生产环境模型不一致时快速切换或者在模型厂商出问题的时候应急换另一家的模型代码一行都不用动。具体到模型选择问数项目的核心是SQL生成和工具调用所以我优先选工具调用Function Calling能力强的模型这一类能力直接决定了Agent能否稳定地使用工具。另外上下文窗口要够大因为要往里塞表结构信息和多轮对话历史。在实操部分我会给出一个最小的模型客户端封装支持从配置里同时读取base_url、api_key、model_name三个参数。2.3 Agent运行时LangGraph环境搭建Agent运行时这一层说白了就是支撑Agent跑起来的代码环境。我用LangGraph作为核心框架所以这块的主要工作是建Python虚拟环境、装依赖、把LangGraph的状态机制用一个最小例子跑通。Python版本建议直接用3.11或3.12。我踩过Python 3.8跑新版本LangChain相关依赖时各种报错的坑提醒一句项目新建直接上3.11省心。这里不是越新越好3.13刚出来的时候不少依赖还没跟上稳一点选3.11或3.12。依赖安装是整个搭建过程里最容易让人崩溃的一步。LangGraph这个生态迭代速度极快包名特别多而且pydantic版本、langchain-core版本经常互相打架。我自己总结了一个原则不要图省事一下装一堆包先按照“跑通最小例子”需要的量来装跑通了再逐步加。核心依赖大概是这几类langgraph本体、langchain-openai提供ChatOpenAI模型封装、openai SDK、pydantic和pydantic-settings配置管理、SQLAlchemy和PyMySQL数据库操作、python-dotenv环境变量加载。具体版本我在实操部分会给一份可以直接用的requirements文件。LangGraph还有一个特殊机制叫Checkpoint检查点它能把Agent每一步的状态保存下来这样支持多轮对话的记忆恢复、人工中断审核、异常恢复重试等功能。基础搭建阶段我们可以先把它的存储用一个本地SQLite或PostgreSQL表搞定不用Redis那么重。后面要做高并发、多实例部署再考虑把存储换成Redis或其他共享存储。3. 动手实操从零把基础设施跑起来3.1 环境初始化与项目目录规范我先定义一个清晰的项目目录。问数项目叫question_agent目录结构如下question_agent/ ├── .env # 环境变量不进仓库 ├── .env.example # 环境变量模板进仓库 ├── .gitignore ├── requirements.txt ├── README.md ├── agent/ # Agent相关代码 │ ├── __init__.py │ ├── state.py # LangGraph状态定义 │ ├── llm.py # 模型客户端封装 │ ├── tools/ # 工具集合 │ │ ├── __init__.py │ │ ├── db.py # 数据库工具 │ │ └── metadata.py # 元数据查询工具 │ └── graph.py # Agent图定义 ├── config/ │ ├── __init__.py │ └── settings.py # 配置类 ├── scripts/ │ ├── init_db.sql # 建表SQL │ └── test_connection.py # 连接测试脚本这个目录不是拍脑袋定的。把代码按agent、config、scripts分层是为了让“模型封装”“工具定义”“流程编排”各归其位后面扩展的时候不用在一个几百行的文件里到处找。很多初学者喜欢把代码全部堆在一个main.py里跑通Demo没问题但项目一复杂就完蛋。接下来是虚拟环境和核心依赖安装cd question_agent python3.11 -m venv venv source venv/bin/activate pip install --upgrade pip pip install langgraph0.2.0 \ langchain-openai0.2.0 \ openai1.40.0 \ pydantic2.7.0 \ pydantic-settings2.3.0 \ python-dotenv1.0.0 \ sqlalchemy2.0.30 \ pymysql1.1.0 \ pandas2.2.0注意版本号我标的是经过验证的下限版本。LangGraph 0.2版本之后API变化较大如果你是照着老教程用StateGraph的旧写法很可能报错。建议直接以新版为准下面我的示例代码就是新版写法。这里还有一个值得说的点为什么要用 virtualenv 而不是 conda 或直接用系统Python。conda也能用但它的包管理有时候会和pip产生冲突尤其LangChain生态有些依赖版本比较敏感。virtualenv加pip的组合最干净出了问题重建环境也快。一个项目一个虚拟环境这是基础设施阶段就要养成的习惯。3.2 配置管理与数据库初始化配置管理这块核心原则就一条代码不掺密钥配置不进仓库。我用.env文件存所有环境变量然后用 pydantic-settings 在代码里统一读取和校验。.env.example文件内容如下你复制一份改成.env填上真实值# 模型配置 LLM_PROVIDERopenai LLM_BASE_URLhttps://api.example.com/v1 LLM_API_KEYsk-your-key-here LLM_MODELgpt-4o-mini LLM_TEMPERATURE0 # 数据库配置 DB_HOST127.0.0.1 DB_PORT3306 DB_USERagent_reader DB_PASSWORDyour-password DB_NAMEquestion_agent_demo DB_ECHOfalse # Agent配置 AGENT_MAX_ITERATIONS10 AGENT_VERBOSEtrue对应配置类写在config/settings.py里from pydantic_settings import BaseSettings, SettingsConfigDict class Settings(BaseSettings): model_config SettingsConfigDict(env_file.env, env_file_encodingutf-8) llm_provider: str openai llm_base_url: str https://api.example.com/v1 llm_api_key: str llm_model: str gpt-4o-mini llm_temperature: float 0.0 db_host: str 127.0.0.1 db_port: int 3306 db_user: str root db_password: str db_name: str question_agent_demo db_echo: bool False agent_max_iterations: int 10 agent_verbose: bool True settings Settings()这里我特别解释一下LLM_TEMPERATURE0这个配置。问数项目里我们要生成SQL、做意图判断这些都是偏确定性任务温度设成0可以让大模型的输出尽量稳定降低随机性带来的SQL差异。我知道有些同学喜欢把温度调高觉得“更有创造性”但在Agent工具调用场景里这是万万要不得的你绝不希望同一个问题两次生成不一样的SQL。数据库初始化方面我先给一个小型的电商销售demo库。为什么用销售数据做demo因为销售这个场景大家都很熟悉订单、用户、产品、区域这些概念不用过多解释方便验证Agent的问答效果。scripts/init_db.sql核心建表语句CREATE DATABASE IF NOT EXISTS question_agent_demo DEFAULT CHARACTER SET utf8mb4; USE question_agent_demo; CREATE TABLE dim_product ( product_id INT PRIMARY KEY AUTO_INCREMENT COMMENT 商品ID, product_name VARCHAR(128) NOT NULL COMMENT 商品名称, category VARCHAR(64) NOT NULL COMMENT 商品类目, unit_price DECIMAL(10,2) NOT NULL COMMENT 单价元, status TINYINT NOT NULL DEFAULT 1 COMMENT 状态1上架0下架 ) COMMENT 商品维度表; CREATE TABLE dim_region ( region_id INT PRIMARY KEY AUTO_INCREMENT COMMENT 区域ID, region_name VARCHAR(64) NOT NULL COMMENT 区域名称, province VARCHAR(64) NOT NULL COMMENT 省份 ) COMMENT 区域维度表; CREATE TABLE fact_order ( order_id BIGINT PRIMARY KEY AUTO_INCREMENT COMMENT 订单ID, order_no VARCHAR(32) NOT NULL COMMENT 订单编号, order_date DATETIME NOT NULL COMMENT 下单时间, product_id INT NOT NULL COMMENT 商品ID, region_id INT NOT NULL COMMENT 区域ID, quantity INT NOT NULL COMMENT 销售数量, amount DECIMAL(12,2) NOT NULL COMMENT 成交金额元, status TINYINT NOT NULL DEFAULT 1 COMMENT 订单状态0已取消1已完成2售后中 ) COMMENT 订单事实表;看着是不是很简单但这里有几个细节值得注意。第一所有表、字段都有中文COMMENT这直接决定了大模型能不能理解表结构是我反复强调的元数据建设基础。第二字段命名用下划线风格虽然拼音乱写也能用但规范命名能减少模型误解。第三金额字段用DECIMAL而不是FLOAT避免浮点精度问题——这种问题一旦发生Agent算出的销售额差了0.1元你排查起来会非常崩溃。建完表之后我还建议写入一部分示例数据。有了数据Agent跑SQL才有结果可返回你才能看到完整的链路效果。示例数据不用多覆盖几个类目、几个区域、近几个月的订单就行具体插入语句这里不展开了你按自己的理解造一批合理的假数据即可。造数据的时候注意日期字段要覆盖最近至少3个月订单状态不要全是已完成要有一些已取消和售后中的这样Agent才能遇到“需要排除已取消订单”这类真实问题。3.3 模型客户端封装模型封装的目标是在你代码的任何一个地方一行代码拿到一个可调用的LLM对象而且切换模型不用改业务代码。agent/llm.pyfrom langchain_openai import ChatOpenAI from config.settings import settings def get_llm(): return ChatOpenAI( modelsettings.llm_model, api_keysettings.llm_api_key, base_urlsettings.llm_base_url, temperaturesettings.llm_temperature, timeout60, max_retries2, )你可能注意到这里用langchain_openai的ChatOpenAI而不是直接调用openai SDK。原因很简单LangGraph生态里的模型节点、工具调用绑定都默认使用LangChain的模型封装格式。直接用OpenAI SDK当然也能调但还得自己处理消息格式转换、工具调用协议解析纯属给自己找麻烦。用ChatOpenAI作为统一入口后续所有LangGraph特性都天然支持。不过千万别把这些写死在代码里所有参数必须走配置。实际开发中经常遇到测试环境用一个模型、生产环境用另一个模型、模型供应商临时出故障要紧急切换的情况配置化之后只需要改.env里的变量然后重启服务。写一个测试脚本验证模型能不能通python -c from agent.llm import get_llm llm get_llm() resp llm.invoke(用一句话介绍你自己) print(resp.content) 如果能正常返回一句话说明模型链路通了。这一步挂了就先解决它不要着急往下走。3.4 数据库连接工具与基础验证有了配置和模型接着写数据库访问层。这里用SQLAlchemy 2.0的engine加连接池应对Agent生成的并发查询足够了。同时考虑到安全建议给Agent创建一个只读数据库账号这非常关键。建一个查询代理账号并授权CREATE USER agent_reader% IDENTIFIED BY your-password; GRANT SELECT ON question_agent_demo.* TO agent_reader%; FLUSH PRIVILEGES;为什么一定要只读账号因为Agent生成SQL是有一定随机性的如果不小心让它拿到写权限一句DELETE FROM fact_order就能把几年的业务数据全干没了。这里不是假设Agent会故意搞破坏而是大模型无法保证100%正确我们必须在权限层面把所有危险操作挡在门外。这个习惯要在一开始就养成。agent/tools/db.py提供一个最简单的SQL执行工具from sqlalchemy import create_engine, text from config.settings import settings engine create_engine( fmysqlpymysql://{settings.db_user}:{settings.db_password}{settings.db_host}:{settings.db_port}/{settings.db_name}?charsetutf8mb4, pool_size5, pool_recycle3600, echosettings.db_echo, ) def run_sql(sql: str) - str: 执行只读SQL返回JSON格式字符串结果。 with engine.connect() as conn: result conn.execute(text(sql)) columns list(result.keys()) rows [dict(zip(columns, r)) for r in result.fetchmany(200)] return json.dumps(rows, ensure_asciiFalse, defaultstr)这里有几个细节大家可能不注意。fetchmany(200)是限制最多取200行防止Agent写了一个没有WHERE条件的全表查询把内存打爆。返回值用JSON字符串是因为LangGraph的State传输要求序列化友好后面接工具返回给大模型也方便。defaultstr是为了处理datetime和Decimal这些不能直接JSON序列化的类型这个坑我印象特别深第一次跑通之前在这里卡了一个多小时。写连接测试脚本# scripts/test_connection.py from agent.tools.db import run_sql if __name__ __main__: result run_sql(SELECT COUNT(*) AS cnt FROM fact_order) print(result)能输出订单总数数据链路就通了。3.5 第一个最小Agent跑通“问题→SQL→结果”链路基础设施全部就位后最后一步是跑一个最小可用的Agent验证整条链路。这一步的核心目标是让LangGraph按流程走一遍“理解问题→生成SQL→执行SQL→总结答案”我们不做复杂的工具调度先把链路打通。LangGraph的新版API实现如下from typing import TypedDict, Literal from langgraph.graph import StateGraph, END from langchain_core.messages import HumanMessage, SystemMessage from agent.llm import get_llm from agent.tools.db import run_sql SYSTEM_PROMPT 你是一个数据分析助手。根据用户的问题生成SQL查询语句。 要求 1. 只查询表 fact_order, dim_product, dim_region 2. 只生成SELECT语句禁止任何修改操作 3. 表字段有中文注释注意理解字段含义 4. 只输出SQL不要解释 class State(TypedDict): question: str sql: str result: str answer: str def generate_sql(state: State) - dict: llm get_llm() resp llm.invoke([ SystemMessage(contentSYSTEM_PROMPT), HumanMessage(contentstate[question]), ]) return {sql: resp.content.strip()} def execute_sql(state: State) - dict: result run_sql(state[sql]) return {result: result} def generate_answer(state: State) - dict: llm get_llm() resp llm.invoke([ SystemMessage(content根据用户问题和查询结果用通俗中文回答用户的问题。如果查询结果为空明确告知查不到数据。), HumanMessage(contentf问题{state[question]}\n查询结果{state[result]}), ]) return {answer: resp.content} app StateGraph(State) app.add_node(generate_sql, generate_sql) app.add_node(execute_sql, execute_sql) app.add_node(generate_answer, generate_answer) app.set_entry_point(generate_sql) app.add_edge(generate_sql, execute_sql) app.add_edge(execute_sql, generate_answer) app.add_edge(generate_answer, END) agent app.compile() if __name__ __main__: result agent.invoke({question: 上个月假设8月卖出多少件商品}) print(SQL:, result[sql]) print(结果:, result[result]) print(回答:, result[answer])这个最小Agent看起来简单但它验证了三件重要的事情模型能按提示词生成SQL、能连上数据库执行查询、能根据结果生成自然语言回答。很多初学者一上来就想做带工具调用的复杂Agent结果链路某一段是坏的根本分不清是哪儿出了问题。先跑通这个最小环相当于给整个系统做了一个冒烟测试后面再往上加工具、加分支、加记忆都是在稳定的地基上盖楼。跑通之后你可以在app.compile()时传入checkpointer参数来启用对话记忆LangGraph会保存每次运行的State这样下一个问题来的时候还能参考上一个SQL的执行经验。基础阶段先用MemorySaver()内存模式就够了后面需要持久化再存到SQLite或Redis。4. 工程化细节日志、异常、安全一个都不能少4.1 日志与可观测性Agent每一步都要留痕Agent开发和传统后端开发有一个非常大的不同传统接口出问题了看报错栈基本能定位Agent出问题往往不是“报错”而是“答错了”或者“绕远了”。比如一次查询结果不对你根本不知道是哪一层出了问题——是意图理解偏了SQL生成错了还是结果解析错了所以日志和可观测性是Agent项目基础设施的重要组成。我建议至少做三件事。第一每个节点入口和出口打日志记录节点名、输入的关键字段、输出的关键字段耗时也顺手记下来。第二打开LangGraph的debug模式它会打印完整的节点调用序列和状态流转。第三生产环境建议接入LangSmith或者自建一套提示词调用日志把每次大模型的输入输出都存下来方便事后复盘。当前基础设施阶段先做到前两点。用logging模块来做配置在入口文件里统一设置import logging logging.basicConfig( levellogging.INFO, format%(asctime)s | %(levelname)s | %(name)s | %(message)s, ) logger logging.getLogger(agent) # 节点里这样用 logger.info(generate_sql start, question%s, state[question])我见过不少项目日志只在出问题时才打印一行except Exception as e: print(e)这对Agent项目是远远不够的。务必让每一步都有迹可循这是排查一切疑难杂症的前提。4.2 异常处理与超时控制高可用不是上线后的事基础设施阶段的另一个工程化任务是异常处理。问数项目最常见的异常有这几类数据库连不上、SQL语法错误、查询超时、模型API限流、Token超限。每一类都应该有对应的处理策略。以SQL执行环节为例我们要区分两类异常。一类是SQL语法错误这种往往是大模型拼错了表名、字段名或语法结构可以在异常里捕获后带着错误信息让模型重新生成一次。一类是执行超时比如大模型写了一个笛卡尔积查询这种重试没意义应该直接返回友好错误并终止。代码层面给run_sql加上超时控制def run_sql(sql: str, max_rows: int 200, timeout: int 10) - str: try: with engine.connect() as conn: conn.execute(text(SET SESSION MAX_EXECUTION_TIME10000)) result conn.execute(text(sql)) ... except Exception as e: return json.dumps({error: str(e)}, ensure_asciiFalse)SET SESSION MAX_EXECUTION_TIME是MySQL端超时如果SQL跑超过10秒就直接终止不浪费数据库资源。对于更严格的场景还可以在数据库账号层面加一条MySQL的max_statement_time限制双重保险。大模型API调用也需要超时和重试。我在get_llm()里已经设了timeout60, max_retries2这个配置基本够用。再往上可以加指数退避的重试策略但基础阶段不必弄得太复杂能保证“失败能感知、不会无限卡死”就够了。4.3 配置安全与密钥管理不要在GitHub上裸奔这一节虽然短但我觉得还是值得单独说一下因为在实际项目里这个问题太常见了。密钥泄漏通常是安全意识问题而不是技术问题。第一.env文件永远不要提交到Git仓库。创建项目那一瞬间就写一个靠谱的.gitignore.env venv/ __pycache__/ *.pyc .DS_Store第二仓库里只放.env.example里面用占位符代替真实密钥。这样新同事拉完代码复制一份.env.example为.env再填自己的配置一分钟就能跑起来而不会把你本地的密钥也一起拉走。第三密钥要有环境隔离。开发、测试、生产环境用不同的.env文件比如.env.dev和.env.prod启动时通过环境变量指定加载哪个。这能防止你在测试环境误操作指向生产数据库。第四如果真的泄漏了别心存侥幸马上到密钥平台吊销并换新没有第二条路。5. 常见问题与排查技巧搭建期最容易踩的坑基础设施搭建阶段的问题大同小异我把问数项目搭建时大家问得最多的几类问题整理成一个速查表都是我自己或朋友实际踩过的。现象可能原因排查方法与解决方案调用模型API报401/403密钥错误或没写入.env文件检查.env是否已加载打印 settings.llm_api_key 确认值确认密钥有无过期模型调用超时base_url填错、网络不通、模型名不存在先用curl直接测API接口确认模型名和厂商平台一致检查timeout配置pydantic版本冲突langchain导入报错依赖版本互相不兼容按requirements.txt版本安装不要无脑装最新版重建虚拟环境连接MySQL报Cant connectMySQL没启动、端口不对、账号权限不对先mysql命令行原生连一次确认排除MySQL自身问题查user表确认账号存在检查bind-addressAgent生成的SQL语法对但语义错元数据不清晰检查表字段COMMENT是否完整确认提示词里给了足够字段说明在提示词中列出常用的字段枚举值SQL执行返回乱码连接串charset没设置utf8mb4连接engine的URL加?charsetutf8mb4确认数据库表本身是utf8mb4LangGraph节点重复执行或死循环State传递异常或边定义错误打开debug模式打印节点顺序检查是否忘记设置END边检查条件边返回值类型返回结果字段缺失SQL执行fetchmany后丢了列类型确认result.keys()能取到列名用defaultstr处理Decimal和datetime展开讲几个我在实践中印象最深的坑。第一个是 pydantic 版本冲突。LangChain生态早期对pydantic v1和v2的兼容性很混乱如果你在一个已经有pydantic v1依赖的旧环境里强行装LangGraph会报各种莫名其妙的属性错误。我自己的做法是项目初始化直接建全新虚拟环境pydantic统一装v2版本依赖用requirements锁定不随意升级。第二个是模型名和API不匹配。很多人用第三方模型服务以为自己填的模型名一定能用。实际上不同的服务商模型名差异很大有的叫gpt-4o有的叫qwen-plus有的叫deepseek-chat填错了虽然HTTP能通但会提示model not found。排查办法就是先用curl或Postman调一次接口别一上来就怪代码。第三个是Agent生成的SQL带着多余内容。大模型有时候会“好心”在SQL前面加一句“好的以下是你要的SQL”或者在结尾加分号和注释。这在提示词里强调“只输出SQL不要解释”能大大缓解但仍有漏网之鱼。更稳妥的办法是在执行前对SQL做一个清理去掉Markdown代码块标记去掉行首行尾的杂质import re def clean_sql(sql: str) - str: sql sql.strip() sql re.sub(r^sql|^|$, , sql).strip() sql re.sub(r;$, , sql) return sql这三行代码看着简单但能把你从大模型的“好心”里救出来。6. 从基础设施到真正可用下一步往哪儿扩基础设施跑通之后问数项目还远不算完。我个人习惯是基础链路一旦通了马上开始迭代“工具层”因为Agent后续的所有能力几乎都建立在工具之上。第一个要加的工具是元数据查询工具。最小值版本可以做一个get_table_schema(table_name)工具让Agent在生成SQL之前先去查一下相关的表结构。这比把所有表结构硬塞到提示词里好得多——一方面省Token另一方面当表特别多、字段特别多的时候只靠系统提示词塞信息模型根本记不住那么多。你可以在第3节最小Agent的基础上把generate_sql节点改造为“先调用元数据工具再生成SQL”。第二个要加的是多轮澄清机制。用户问题模糊的时候比如“最近销售怎么样”Agent应该主动反问“你是指销售额还是销量最近是指近7天还是近30天”这个能力需要Agent具有多轮对话的记忆能力也就是在第3节提到的checkpointer机制的基础上实现。第三个值得关注的是MCP协议。MCPModel Context Protocol如果你还没接触过可以把它理解成AI应用里的“USB接口标准”——你按照这个协议封装数据源和工具任何支持MCP的Agent客户端都能直接即插即用。问数项目里的数据库查询工具、元数据工具都可以考虑封装成MCP服务这样未来如果要把同一个能力复用到其他Agent项目里就不用重写一遍。这个方向我已经在探索了后面专门写一篇。还有一点是很现实的如果问数项目要接给业务团队用你肯定不希望业务同学每次问完都等十几秒。系统延迟优化要靠缓存和异步把高频问题对应的查询结果缓存到Redis遇到相同或相似的问题直接命中缓存。这一步属于性能基础设施的范畴等并发量真的起来了再做也不迟。说到最后我再分享一点自己的真实体会。基础设施搭建这件事看起来没有写Agent逻辑那么“高级”但它的好坏直接决定了你后期开发的幸福感。我见过太多项目Demo阶段跑得飞起一到接真实数据库、接真实业务场景就各种崩根本原因就是基础没打牢——配置写死、没有日志、权限混乱、依赖一堆坑。反过来如果你愿意在这层多花点心思把配置、目录、日志、安全、异常处理一次做到位后续所有Agent能力的迭代都会非常顺畅。搭建过程里如果遇到问题优先怀疑版本和配置这两个是最大的坑源。把这一篇的内容都跑通了咱们下一篇就可以正式开搞Agent的核心逻辑——工具调用、状态编排、多轮对话。一个真正能用的问数项目已经离你不远了。
