Langchain多智能体架构驱动的数据检索与可视化系统实现

简介本资源是一个基于Python实现的Langchain多智能体数据检索与可视化系统面向数据分析工程师、AI应用开发者及高校相关专业学习者解决多源异构数据高效检索、协同分析与交互式可视化呈现的核心问题。压缩包共14个文件含4个核心Python脚本如main.py、graph.py、display.py等分别承担主控调度、知识图谱构建与前端渲染、8张界面与流程示意图PNG以及requirements.txt和README.md文档整体大小5.68MB结构清晰、开箱即用。已有58人学习下载资源完整覆盖多智能体协作架构设计、Langchain链式调用实现、Streamlit动态可视化集成等关键技术环节附带可运行代码、依赖清单与说明文档便于快速部署、调试与二次开发。1. 项目概述与设计动机1.1 这个项目到底解决了什么问题先说结论这是一个把查数据和看数据两件原本割裂的事情通过 Langchain 多智能体架构串成一条自动化流水线的系统。用户只需要用自然语言提问比如帮我查一下今年华南区各产品线的销售趋势系统会自动完成意图解析、数据检索、指标计算、图表生成四步工作最终输出一张可以直接用的可视化图表和对应的数据摘要。你可能会问直接用 Pandas Matplotlib 写个脚本不也能做吗这里面的差异在于检索二字。传统方案里数据源是固定的 CSV 或数据库表查询逻辑是写死的而这个项目面向的是多数据源、动态表结构、甚至包含非结构化文档的场景。用户的问题千变万化表结构可能今天加一列明天减一列指标口径也可能随业务调整。这时候靠硬编码规则根本扛不住所以才需要让大模型驱动的 Agent 去理解数据字典、规划查询路径、调用合适的工具链。从 zip 包的名字来看这显然是一份可以直接交付的工程化项目而不是教学 demo。实际解压后我估计目录里应该包含后端服务大概率是 Flask 或 FastAPI、Langchain 智能体编排层、数据处理模块、前端可视化页面以及一份环境配置文件。这类项目的价值在于它是一个可以演示、可以二次开发、也可以直接嵌入到现有业务系统里的完整闭环。1.2 为什么选多智能体而不是单 Agent这是整个项目里最值得拆解的设计决策。很多人刚开始接触 Langchain 时习惯用一个 Agent 挂一堆工具凑合着用。但对于数据检索和可视化这个场景我强烈建议拆成多个 Agent原因有三条。第一工具上下文隔离。一个 Agent 如果同时挂数据库查询工具、文档检索工具、代码生成工具、图表渲染工具它的 System Prompt 会变得非常臃肿而且每次推理都要把所有工具的描述塞进上下文里既浪费 token 又容易让模型混淆该调用哪个。拆分之后每个 Agent 只需要关心自己管辖范围内的小工具集决策清晰度会高很多。第二职责边界决定了提示词质量。你的数据检索 Agent 需要理解 SQL 方言、表结构、字段含义你的可视化 Agent 需要理解 ECharts 的配置项、色板规范、图表类型适用场景。这两者的专业术语完全不同强行塞进同一个 Prompt 里模型在切换任务时容易产生角色遗忘。多智能体架构天然要求你为每个角色写独立的 System Prompt反而倒逼你更仔细地设计指令。第三容错与重试策略可以分而治之。数据库查询失败只需要重试查询 Agent图表配置报错只需重新调度可视化 Agent。如果单 Agent 一把梭任何一环出错都可能让整个链路重新跑一遍资源浪费和时间开销都很可观。从热词里能看到大量关于 Langchain 和 LangGraph 的对比讨论这其实是同一个问题的两面。Langchain 原生的 AgentExecutor 是扁平化的单次决策循环适合简单任务LangGraph 则允许你定义带环路的图结构天然适配多智能体协作。这个项目既然是多智能体系统采用 LangGraph 或者基于 Langchain 的 Agent 组合模式来编排会比单纯用 AgentExecutor 嵌套更合理。2. 系统架构与核心模块设计2.1 整体架构一条从问句到图表的流水线我先把这套系统的模块拆解出来你可以对照着看后面每一块我都会展开讲。模块职责关键技术点输入输出用户入口层接收自然语言问题Flask/FastAPI 路由、WebSocket文本 - JSON 请求意图识别 Agent判断任务类型、拆分多步子问题LLM 分类、少样本示例问题 - 子任务列表数据源管理模块维护数据源注册信息与连接SQLAlchemy / 连接池 / 元数据缓存表名 - 可查询的数据源数据检索 Agent生成并执行数据查询Text2SQL、RAG、工具调用子任务 - DataFrame / JSON可视化 Agent选择图表类型并生成配置ECharts 配置生成、代码解释器数据 - 图表配置 渲染代码结果整合 Agent汇总多路结果生成最终答复Prompt 模板、内容摘要多段结果 - 最终答案前端展示层渲染对话和图表Vue/React 或纯 HTML 模板JSON - 交互界面如果从数据流的角度看一次完整的请求链路是这样的用户输入问题首先进入意图识别模块由它判断这个问题需要查数据库、查文档还是不需要查数据直接回答。如果需要查数据就进入数据检索环节检索结果交给可视化模块判断是否适合出图最后所有输出汇总成一致的回复结构返回到前端。这套架构里最容易被忽略但实际最影响体验的是数据源管理模块。没有它检索 Agent 就是个瞎子——它不知道有哪些表、哪些字段、数据在哪个库。所以现在的主流做法是维护一份数据字典包含表名、字段名、字段类型、字段注释、示例值通过 Embedding 的方式灌入向量库。检索时先从向量库召回相关表结构再把这些结构信息作为上下文拼接进 Text2SQL 的 Prompt 里。2.2 为什么选 LangGraph 作为编排框架我把 Langchain 和 LangGraph 的差异放一张表里你就能直观理解选型逻辑维度Langchain AgentExecutorLangGraph任务形态单 Agent 循环多节点图编排流程控制简单循环条件退出条件边、并行节点、循环、子图状态管理靠内存变量传递显式 State 对象可观测性主要在回调日志每一步状态可序列化查看适用场景简历筛选、代码生成等单步决策多角色协作、需要人工介入的审批流这个项目明显属于后者。数据检索 Agent 可能需要在数据库查询工具和文档检索工具之间做路由可视化 Agent 需要在生成配置后做一次自检如果配置非法需要自动修正。这些流程不是简单的多工具循环能覆盖的更像是流程图。LangGraph 把每个 Agent 定义成图的节点节点之间通过条件边连接整体流程一目了然出问题也容易定位。如果你最终拿到手的项目用的是老式 Langchain AgentExecutor 组合也不用慌后面我会讲如何改造成 LangGraph 的版本。但无论是哪种写法核心设计思想是一致的一个 Planner 负责拆解任务多个 Executor 各管一段一个 Evaluator 负责检查结果质量。2.3 多智能体的角色定义与 Prompt 设计思路既然是多智能体系统每个 Agent 的人设就决定了系统的最终表现。我按实际项目中比较成熟的角色划分方式展开讲讲你可以直接在代码里对照。第一个是 Planner Agent。它的 System Prompt 核心内容是把用户输入拆解成可执行的子任务列表每个子任务包含任务描述、依赖的数据源、期望输出格式。这里要用到 few-shot 示例因为 LLM 对不同类型的问句表现差异很大。比如2023年和2024年各季度销售额对比这样的问题Planer 应该拆出两个查询子任务加一个对比可视化子任务。第二个是数据检索 Agent。它是整个系统里 Prompt 设计最讲究的一个。核心词有两类一类是数据库操作指令一类是领域专用词汇定义。如果数据表里有个字段叫 ARPU你不告诉模型它是什么模型可能不会正确使用所以必须把字段字典灌进上下文。有了这些信息之后模型负责把自然语言翻译成 SQL查询执行器负责实际跑 SQL然后把结果集转成结构化数据。第三个是可视化 Agent。我见过很多失败的案例都是让模型直接输出一段完整的 ECharts 配置结果不是坐标轴标签乱码就是图表类型选择完全不合理。比较好的做法是两步走第一步让模型输出结构化描述包括图表类型、X轴和Y轴字段、聚合方式、颜色偏好第二步把结构化描述渲染成前端可执行的配置代码。这样一步出配置一步出代码错误率会大幅下降。第四个是 Critic Agent很多人容易漏掉这一层但它非常关键。它的职责是给可视化结果和查询结果打分判断结果是否合理。比如查询结果为空Critic 就需要分析是 SQL 写错、过滤条件过严、还是表里真的没数据然后决定是重新生成还是向用户说明情况。这几个角色之间的接口边界一定要定义清楚。最简单的约定是所有 Agent 之间通过 JSON 传递信息每一条 JSON 至少包含 role、content、structured_data 三个字段。这样在后面接入任何编排框架时都可以无缝对接。2.4 zip 包里应该包含的目录结构一探究竟拿到 zip 包先别急着跑起来先看看目录结构。一个规范的项目目录结构本身就暴露了很多设计信息。按照行业实践我推测这份项目的目录应该长这样data_agent/ ├── agents/ # 各智能体实现 │ ├── planner.py │ ├── query_agent.py │ ├── visualize_agent.py │ └── critic_agent.py ├── core/ # 核心配置与状态管理 │ ├── config.py │ ├── state.py │ └── llm_factory.py ├── datasource/ # 数据源适配层 │ ├── base.py │ ├── mysql_source.py │ └── csv_source.py ├── tools/ # Langchain 工具封装 │ ├── sql_executor.py │ ├── chart_configurator.py │ └── doc_retriever.py ├── web/ # 前端与接口层 │ ├── app.py │ ├── static/ │ └── templates/ ├── requirements.txt └── README.md记住一句话如果一个项目连 agents 目录都没有就改名叫Langchain 单 Agent 数据查询项目比较诚实。目录里如果有 docker-compose.yml说明作者考虑了环境一致性如果没有大概率需要你自己配环境后面我会详细讲配置步骤。3. 数据检索核心链路如何让 Agent 理解数据并查得准3.1 数据源接入与元数据构建这一节是整个系统最容易被低估的部分。Text2SQL 的本质不是大模型懂 SQL而是大模型在足够上下文信息的辅助下写出正确 SQL。上下文的核心就是元数据。构建元数据的方法很简单就是遍历数据源里所有表提取每个字段的名称、类型、注释、枚举值、示例数据然后做三件事把表名和字段名的自然语言描述做 Embedding灌入向量库供检索把完整的数据字典转成带格式的文本作为 Prompt 库供拼接把表间关系外键、常用 JOIN 路径单独记录辅助多表查询比如 MySQL 侧可以执行一条语句拿到所有字段信息SELECT TABLE_NAME, COLUMN_NAME, DATA_TYPE, COLUMN_COMMENT FROM INFORMATION_SCHEMA.COLUMNS WHERE TABLE_SCHEMA your_database;然后在 Python 里处理成 Langchain 需要的格式metadata_docs [] for row in cursor.fetchall(): table_name, col_name, col_type, col_comment row doc f表{table_name}的字段{col_name}类型{col_type}含义{col_comment} metadata_docs.append(doc) from langchain_community.vectorstores import FAISS from langchain_openai import OpenAIEmbeddings vectorstore FAISS.from_texts(metadata_docs, OpenAIEmbeddings()) retriever vectorstore.as_retriever(search_kwargs{k: 8})这里我强烈建议对字段描述做一次人工审核。很多数据库的字段注释是开发随便写的比如a、b、remark1这种注释直接丢给大模型等于给模型喂噪声。遇到这种情况宁可自己写一份字段补充说明文档也不要让模型去猜。3.2 Text2SQL 的实现细节与查询执行器封装Text2SQL 是整个数据检索链路的核心环节。Langchain 官方生态里有 create_sql_query_chain 可以直接用但真实项目里我基本不用它因为它生成的 SQL 只返回字符串执行还得自己做。我更倾向于把生成 SQL、检查 SQL、执行 SQL合并到一个工具函数里。核心代码如下from langchain_openai import ChatOpenAI from langchain_core.prompts import ChatPromptTemplate sql_prompt ChatPromptTemplate.from_messages([ (system, 你是一个数据分析专家。根据数据字典和用户问题生成一条 MySQL SQL。 要求\n 1. 只返回 SQL 本身不要多余解释\n 2. 如果没有匹配的表回答 NO_SCHEMA\n 3. 注意处理空值\n 4. 时间字段统一为 YYYY-MM-DD 格式\n 数据字典\n{schema}\n), (human, 用户问题{question}) ]) llm ChatOpenAI(modelgpt-4o-mini, temperature0) query_chain sql_prompt | llm def execute_query(question: str, schema: str, conn) - dict: sql query_chain.invoke({schema: schema, question: question}).content # 安全检查禁止非 SELECT if not sql.strip().upper().startswith(SELECT): return {error: 只允许执行 SELECT 查询} # 执行 df pd.read_sql(sql, conn) return {sql: sql, data: df.head(200).to_dict(orientrecords)}这里有几个坑我必须提醒你第一必须加 SQL 安全检查。Agent 偶尔会生成 DELETE、UPDATE 甚至 DROP 语句虽然概率不高但一旦发生就是灾难。最简单的做法是白名单校验只放行 SELECT。第二必须限制返回行数。大模型写的 SQL 经常忘了加 LIMIT遇到千万级数据的表直接 OOM。正确做法是把原始 SQL 包一层子查询统一加 limit而不是等模型自觉。第三结果字段名要标准化。数据库里字段可能是 create_time前端习惯用 createTime如果你不在查询层做一次字段重命名后续可视化 Agent 生成图表配置时很容易因为字段对不上而报错。3.3 RAG 与结构化查询的混合检索策略数据检索不只有数据库这一种形态。项目标题里的数据检索在很多真实场景里包含了一部分是查数据库一部分是查文档还有一部分是查知识库。所以这套系统最好支持混合检索模式。我之前做过一个运维知识助手用户问磁盘告警的排查步骤有哪些这明显不是数据库能回答的需要去查运维手册但如果用户问最近一个月磁盘使用率超过80%的主机有多少台这就要查监控数据库。两种问题混在一个系统里单 Agent 很难判断用哪个工具多智能体架构的价值在这里就体现出来了意图识别 Agent 先判断是结构化查询还是非结构化检索如果是结构化查询走 SQL 工具链如果是非结构化检索走向量检索工具链如果问题模棱两可把两种结果都召回用 Reranker 做排序再交给最终回答向量检索部分的核心配置from langchain_community.vectorstores import Chroma from langchain_openai import OpenAIEmbeddings embeddings OpenAIEmbeddings(modeltext-embedding-3-small) vectorstore Chroma( persist_directory./chroma_db, embedding_functionembeddings ) retriever vectorstore.as_retriever( search_typemmr, # 最大边际相关性避免召回结果太相似 search_kwargs{k: 6, lambda_mult: 0.7} )混合检索里一个容易被忽视的参数是召回条数。太多会稀释有效信息太少可能漏掉关键内容。我做过的大多数项目里k 取 4-8 是比较合理的区间。如果你用的是多路召回加融合排序每路召回的 k 可以更小但是路数可以多一些。4. 可视化 Agent 的工程实现4.1 从数据到图表类型的选择逻辑可视化 Agent 最常见的失败模式是用户明明要一个趋势图它却搞了一个饼图。这可能不是模型能力问题是你在 Prompt 里没有把图表类型的选择规则写清楚。我总结了一套选择规则可以直接写进系统提示词用户意图推荐图表备注随时间变化趋势折线图/面积图X 轴必须是时间字段分类对比柱状图/条形图类别超过 8 个时转横向构成占比饼图/环形图/堆叠柱状图分类不超过 7 个两个变量相关性散点图数据点少于 200地理分布地图需要经纬度或地区名称数据分布箱线图/直方图关注分位数这套规则放进 System Prompt 后模型很少再犯低级的选择错误。你还可以在 Prompt 里加一句如果存在更合适的图表类型请解释为什么不用默认推荐的类型引导模型展示推理过程。下面这段是我在实际项目里验证过的可视化配置生成代码片段analysis_prompt ChatPromptTemplate.from_messages([ (system, 你是一位资深数据分析师。根据数据样例和用户的展示需求输出 JSON 格式的可视化配置。\n 可用图表类型line, bar, pie, scatter, area\n 字段必须来自给定的数据样例列名。\n 输出格式示例\n {{\chart_type\: \bar\, \x_field\: \month\, \y_field\: \sales\, \title\: \月度销售额\}}\n), (human, 数据样例{sample}\n需求{requirement}) ]) def generate_chart_config(data: list[dict], requirement: str) - dict: sample json.dumps(data[:5], ensure_asciiFalse) result (analysis_prompt | llm).invoke({ sample: sample, requirement: requirement }) # 从回复中提取 JSON推荐用 json.loads 并做异常兜底 return extract_json(result.content)4.2 ECharts 配置生成与前端渲染细节当 Agent 输出了图表类型和字段映射之后下一步是把它们翻译成 ECharts 的 option 配置对象。这一步有两种做法一种是让大模型直接生成完整 option JSON另一种是用模板拼接。我推荐后者因为 ECharts 有大量的可选配置项完全交给模型生成容易漏掉关键属性。我的做法是定义一个从结构化描述到 ECharts option 的转换器def build_echarts_option(config: dict, data: list[dict]) - dict: chart_type config[chart_type] x_field config[x_field] y_field config[y_field] # 构造基础 option option { title: {text: config.get(title, )}, tooltip: {trigger: axis}, xAxis: {type: category, data: [d[x_field] for d in data]}, yAxis: {type: value}, series: [{ name: y_field, type: chart_type, data: [d[y_field] for d in data] }] } # 如果是饼图需要做一次 transform if chart_type pie: option[xAxis] None option[yAxis] None option[series][0] { type: pie, radius: 60%, data: [{name: d[x_field], value: d[y_field]} for d in data] } return option为什么不在这一步用大模型因为 option 的构造是确定性逻辑大模型生成 JSON 偶尔会多一个逗号少一个引号出了问题排查成本很高。模板方案快、稳、可控。如果后续想支持更多图表类型往模板里加 case 分支即可。前端渲染部分如果项目用的是 Flask最简单的做法是用一个基线模板页面通过 Jinja2 传递 option 到前端脚本里fetch(/api/chat, { method: POST, headers: {Content-Type: application/json}, body: JSON.stringify({question: userInput}) }) .then(res res.json()) .then(res { if (res.chart_option) { const chart echarts.init(document.getElementById(chart)); chart.setOption(res.chart_option); } if (res.answer) { document.getElementById(answer).innerText res.answer; } });5. 前端集成与后端接口设计5.1 Flask 多智能体后端的接口约定既然包名里有 zip说明作者交付的应该是一份可运行的完整项目前后端大概率是集成的。这里我以 Flask 为例写一下接口设计的思路。后端只需要暴露两个核心接口一个是对话接口接收用户问题返回答案、图表配置和中间状态另一个是数据源管理接口用于维护数据源和刷新元数据。app.route(/api/chat, methods[POST]) def chat(): data request.get_json() question data.get(question, ) # 1. 规划阶段 plan planner_agent.run(question) # 2. 数据检索阶段 result query_agent.run(plan) # 3. 可视化阶段 if result.get(data): chart_option visualize_agent.run(result[data], plan.get(visual_requirement)) # 4. 汇总输出 final_answer result_agent.run({ question: question, sql: result.get(sql), answer: result.get(summary), }) return jsonify({ answer: final_answer, chart_option: chart_option, sql: result.get(sql), })接口设计里有几个小细节值得注意一是响应时间控制。LLM 推理本身耗时大加上数据库查询和前端渲染整个链路可能超过 10 秒。前端一定要有 loading 状态建议在后端加一个简单的异步任务机制比如提交任务后立刻返回 task_id前端轮询获取结果。虽然实现上要复杂一些但体验差距很大。二是中间状态透传。多智能体系统的一个好处是你能拿到每一层的中间结果包括规划结果、SQL 语句、数据条数等等。把这些信息透传给前端做成调试面板对开发排查问题非常有帮助。用户可能不关心这些但你自己爽。5.2 前端页面的布局与交互设计页面布局上我建议做成经典的左侧对话流 右侧图表区双栏结构。对话流里展示用户提问和系统回答回答里包含文字摘要和对应的图表卡片右侧区域放大展示最近一张图表支持下载、全屏预览。前端不要过度设计这里功能优先级排序是输入框可多行输入、支持回车发送对话气泡可区分用户和系统图表卡片随回答流插入每次新查询自动滚动到最新消息如果项目里前端部分比较弱也可以直接用 Plotly 或者 PyG2 生成 HTML 片段然后通过 iframe 嵌进来。这样后端可以全包前端开发成本基本为零适合快速演示。6. 核心 Agent Prompt 与工具链配置6.1 各角色 Prompt 设计要点下面我按角色把 Prompt 设计的核心要点拆一遍每一条都是踩过坑之后总结出来的。Planner Agent 的 Prompt 核心要素要求输出结构化 plan每个子任务必须有明确的依赖关系使用 few-shot 示例至少给 3 个不同类型的问题样例明确声明做不到的事情直接说做不到不要强行拆子任务指定输出为 JSON 数组字段包含 task_id、description、dependencies、expected_outputQuery Agent 的 Prompt 核心要素明确数据字典的格式和使用方式告诉模型优先使用注释字段来理解业务含义给出一组好 SQL 示例和坏 SQL 示例要求模型执行查询前先在内部确认表是否存在、字段是否匹配输出时必须附带查询用的 SQL供审计和调试Visualize Agent 的 Prompt 核心要素明确图表类型选择的判断路径强调字段必须来自数据结果集不允许虚构字段要求在配置生成后做一次检查确认 X 轴字段不是数值型但被当成了连续轴或者字段名拼写错误Critic Agent 的 Prompt 核心要素审查 SQL 结果的数据类型是否符合预期判断查询结果是否为空并给出可能原因检查图表配置的 data 字段是否与 review 数据一致输出 pass/fail 修改建议反馈给对应的 Agent6.2 Langchain 工具链封装从函数到 ToolLangchain 里创建工具非常直接你可以用 tool 装饰器把普通函数变成工具from langchain_core.tools import tool tool def query_data(sql: str) - list: 执行数据库查询并返回结果列表。输入必须是 SELECT 开头的 SQL 语句。 if not sql.strip().upper().startswith(SELECT): raise ValueError(只支持 SELECT 查询) df pd.read_sql(sql, engine) return df.head(100).to_dict(orientrecords) tool def generate_chart(chart_type: str, x_field: str, y_field: str, data_json: str) - str: 生成 ECharts 配置返回 JSON 字符串。 data json.loads(data_json) option build_echarts_option({ chart_type: chart_type, x_field: x_field, y_field: y_field }, data) return json.dumps(option, ensure_asciiFalse)工具描述是 Langchain 工具里最容易被忽略但最有价值的部分。好的工具描述应当做到让模型读到描述时就能清楚知道什么情况下调用这个工具、输入参数各代表什么、有什么注意事项。另一个讲究是工具粒度。工具太粗模型不知道该传什么参数工具太细工具列表过长影响推理速度和准确性。我个人的经验是一个 Agent 挂在 4-6 个工具以内每个工具职责单一描述不超过 100 字。6.3 环境配置与依赖安装这个项目你需要准备的环境Python 3.10 或 3.113.9 以下跑不了新版 Langchain一个可用的 LLM APIOpenAI 兼容接口即可MySQL 或其他数据库提前准备好测试的表数据Node.js如果前端用了工程化构建不需要则忽略依赖安装建议直接看 requirements.txt没有的话手动装这几样pip install langchain langchain-openai langgraph fastapi uvicorn pandas sqlalchemy pymysql flask flask-cors openai tiktoken chromadb安装完成后写一个最简单的连通性测试from langchain_openai import ChatOpenAI llm ChatOpenAI( modelgpt-4o-mini, api_keyyour-api-key, base_urlhttps://api.openai.com/v1 ) resp llm.invoke(你好) print(resp.content)如果这条通了说明 Langchain 到 LLM 的通路没问题后面出问题就大概率出在数据源或业务流程上。7. 实操过程与核心环节实现7.1 快速跑起这个项目分步说明假设你现在已经解压了 zip 包我按最稳妥的顺序带你过一遍第 1 步准备虚拟环境cd data_agent python -m venv venv source venv/bin/activate # Windows 用 venv\Scripts\activate pip install -r requirements.txt如果 requirements.txt 不存在就用我前面列的那一串安装命令。第 2 步配置环境变量在项目根目录创建.env文件OPENAI_API_KEYsk-xxxx OPENAI_API_BASEhttps://api.openai.com/v1 DB_HOSTlocalhost DB_PORT3306 DB_USERroot DB_PASSWORDyourpassword DB_NAMEtest_db注意如果你没有 OpenAI 的 key可以用任何兼容 OpenAI 接口的国产模型服务只需要改 base_url 和 model 名称。第 3 步构建数据字典这一步非常关键执行数据字典构建脚本python scripts/build_metadata.py脚本会扫描配置的数据库生成表结构文档并灌入向量库。执行完你会在项目目录下看到多了一个metadata_index/文件夹里面就是向量索引文件。第 4 步启动后端服务python web/app.py看到Running on http://127.0.0.1:5000就说明启动成功。用浏览器打开这个地址你应该能看到一个对话页面。第 5 步测试对话在输入框里输入找出销售额最高的 5 个产品并按销量排序用柱状图展示。如果一切正常你会看到回答中有 SQL 展示、柱状图渲染和文字摘要。7.2 核心链路手把手演示一次完整的查询过程假设你的数据库里有sales_records表字段包括product_name、sales_amount、sales_qty、sale_date、region。用户输入的问题是帮我分析 2024 年华北区各月的销售趋势。系统内部的执行流程如下第一步Planner Agent 收到请求输出[ { task_id: 1, description: 查询 2024 年华北区各月销售额, deps: [], expected_output: 月份和销售额的二维数据 }, { task_id: 2, description: 生成趋势图展示, deps: [1], expected_output: ECharts 折线图配置 } ]第二步Query Agent 拿到 task_id 1 的描述先从元数据向量库中检索出sales_records表的结构然后生成 SQLSELECT DATE_FORMAT(sale_date, %Y-%m) AS month, SUM(sales_amount) AS total_sales FROM sales_records WHERE region 华北区 AND sale_date 2024-01-01 AND sale_date 2025-01-01 GROUP BY month ORDER BY month;执行后得到 12 行月度数据。第三步Visualize Agent 拿到数据根据趋势这个语义选择折线图并生成配置{ chart_type: line, x_field: month, y_field: total_sales, title: 2024年华北区月度销售趋势 }第四步Critic Agent 审核配置确认月份落在 X 轴、销售额落在 Y 轴、图表类型正确返回 pass。最终前端收到响应渲染折线图和文字摘要。7.3 从 AgentExecutor 迁移到 LangGraph 的改造思路如果你的项目里用的是老式 AgentExecutor你可以考虑把它升级成 LangGraph 的版本。LangGraph 的核心理念是图节点就是各个处理函数边定义了执行顺序和条件。我这里给一个最精简的改造示例from langgraph.graph import StateGraph, END class AgentState(TypedDict): question: str plan: list sql: str data: list chart_option: dict final_answer: str def planner_node(state: AgentState): plan planner_agent.run(state[question]) return {plan: plan} def query_node(state: AgentState): result query_agent.run(state[plan][0]) return {sql: result[sql], data: result[data]} def visualize_node(state: AgentState): option visualize_agent.run(state[data], state[plan][1]) return {chart_option: option} def answer_node(state: AgentState): answer result_agent.run(state) return {final_answer: answer} # 构建图 graph StateGraph(AgentState) graph.add_node(planner, planner_node) graph.add_node(query, query_node) graph.add_node(visualize, visualize_node) graph.add_node(answer, answer_node) graph.set_entry_point(planner) graph.add_edge(planner, query) graph.add_edge(query, visualize) graph.add_edge(visualize, answer) graph.add_edge(answer, END) app graph.compile()LangGraph 的好处是你可以很方便地加入条件分支。比如 query_node 查询失败时可以重新调用 planner 而不是继续往下走def should_retry(state: AgentState) - str: if state.get(error): return planner return visualize graph.add_conditional_edges(query, should_retry, {planner: planner, visualize: visualize})这个迁移的收益是可观测性和可控性大幅提升缺点是要重写一部分胶水代码。如果你的项目规模不大AgentExecutor 组合也够用可以先跑通再考虑迁移。8. 常见问题与排查技巧实录8.1 LLM 相关报错与对策问题 1调用模型 API 报认证失败这个最常见检查三个地方环境变量是否加载成功、key 是否过期、API base 地址是否填对。在 Python 里可以用os.getenv直接打印看环境变量是否真的取到了。import os from dotenv import load_dotenv load_dotenv() print(os.getenv(OPENAI_API_KEY)[:8] ****)问题 2模型输出不是合法 JSON多智能体之间传递状态几乎都依赖 JSON而 LLM 输出偶尔会夹杂 Markdown 代码块标迹或多余解释。一个通用的兜底方案是提取 JSON 片段后做一次清洗import re, json def extract_json(text: str): match re.search(r\{.*\}|\[.*\], text, re.DOTALL) if match: try: return json.loads(match.group()) except json.JSONDecodeError: # 做一次括号补全重试简易实现 return repair_json(match.group()) raise ValueError(无法从模型输出中提取 JSON)问题 3上下文过长导致 token 超限数据字典拼接起来经常超长。解决方案有几个一是精简元数据只把和当前问题相关的表放进来二是用摘要代替全文三是按字段重要程度排序优先放高价值的表结构。还有一个容易被忽略的点SQL 执行结果超过一定条数时要截断不要一股脑全塞给可视化 Agent。8.2 数据检索与 SQL 生成问题问题 4生成的 SQL 语法错误或字段不存在排查思路是把 Query Agent 生成的 SQL 打印保存然后在数据库客户端手动执行一遍判断是模型生成错误还是执行环境的问题。如果是模型生成错误通常原因是元数据缺失。比如字段注释不清晰、表名不是通俗命名、数据库大小写敏感设置特殊。解决方法是完善元数据或者在 Prompt 里增加拼接表名前必须验证表名存在的约束。问题 5查询结果为空但业务上应该有数据这个很常见原因可能是时间过滤条件写错、地区名称不匹配比如库里存的是华北用户问的是华北区、空格或大小写问题。可以在 Prompt 里增加一条规则当查询结果为空时去掉所有过滤条件重跑一次看看数据总量是否正常然后再逐步加回条件定位问题。8.3 可视化与前端问题问题 6图表不显示或空白优先打开浏览器控制台看报错。最常见的两类错误一是前后端字段名不一致前端取data.company后端返回的 key 是company_name二是 ECharts 的 option 里混入了NaN或undefined导致渲染失败。问题 7图表类型乱选例如2023年各季度销售情况模型可能用柱状图这个其实也能接受但如果用户明确要趋势它选了饼图那就是 Prompt 里图表选择规则不够明确。把图表选择规则表直接放进 System Prompt并加上当用户提到趋势/走势/变化时必须选择折线图这种硬性约束。8.4 性能与资源问题问题 8响应时间太长数据检索系统的瓶颈通常在三个地方LLM 推理耗时、数据库查询耗时、上下文太长导致 LLM 处理慢。排查顺序是先看接口日志确认时间消耗在哪一环。如果是 LLM 推理慢可以换更快的模型如果是数据库慢给常用查询建索引如果是上下文太长精简 Prompt 和元数据。问题 9多用户并发时线程安全Flask 默认开发服务器不擅长并发。如果你临时用 Flask 的 dev server 跑碰到多个用户同时提问会卡死。改用 gunicorn 加多 worker 启动gunicorn -w 4 -b 0.0.0.0:5000 web.app:app如果每个 worker 里都要加载 LLM 客户端和向量库内存开销会比较大建议把 LLM 和向量库做成模块级单例避免重复初始化。9. 扩展方向与实用建议9.1 从单轮到多轮对话的演进目前的系统如果做成多轮对话需要处理一个关键问题上下文记忆。用户先问2024 年华北区销售额再问同比增长多少后者依赖前者的查询上下文。要支持这种对话需要在状态对象里维护一个当前数据上下文也就是把上一次查询结果的 schema 和数据摘要存下来作为下一轮查询的参考。LangGraph 的 State 机制天然支持这个需求。你只需要在 AgentState 里加一个history字段每一轮结束后把当前问题、SQL、结果摘要追加进去。Query Agent 在生成 SQL 前会先读取历史理解用户指代关系。9.2 接入更多数据源和实时数据当前系统如果只支持 MySQL 和 CSV可以扩展支持 ClickHouse、PostgreSQL、MongoDB、API 接口、甚至是实时流数据。接入新数据源的关键是写好适配器统一输出 DataFrame 格式这样上层 Agent 完全不需要感知数据源差异。如果是实时数据场景可以加一个定时刷新任务每隔一段时间重新拉取数据并更新状态更高级的做法是接消息队列数据变化时主动推送事件。9.3 加入报表自动生成与定时推送这套系统扩展成智能报表平台也是一条不错的路线。用户下发一次查询需求后系统每周定时执行一遍把结果和图表发送到指定邮箱或 IM 群。实现方式很简单写一个定时任务脚本调用系统内部的核心函数然后把结果渲染成 HTML 或图片再发送。9.4 最后分享一个我在实操中的体会多智能体系统的核心竞争力不在模型而在你对业务的理解和流程的设计。模型谁都拿得到但你的数据字典、你的图表类型选择规则、你的异常处理策略这些才是项目的护城河。拿这次解包的实战来说最花时间的地方不是 Langchain API 的调用而是把数据库里那些语焉不详的字段注释一条条补充完整把各种边角情况空值、时区、单位不一致在 Prompt 规则里明确写清楚。这些工作看似枯燥但正因为做了这些系统在真实数据上才能稳定输出也才真正算得上一个能交付的项目。如果你正准备照着这个 zip 包搭一套自己的系统我的建议是第一步先不带 Agent直接用硬编码流程把查数据 - 出图表这条主链路跑通第二步再引入多 Agent把每个环节替换成智能体第三步再上 LangGraph 优化编排。每一步的验证成本都低出了问题也容易定位。这样走下来你得到的不仅是一个能跑的系统还会对每一层背后的原理有更深的体感。本文还有配套的精品资源点击获取