企业官网建设流程全解析

简介：本资源是一个基于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 整体架构：一条从问句到图表的流水线

我先把这套系统的模块拆解出来，你可以对照着看，后面每一块我都会展开讲。

如果从数据流的角度看，一次完整的请求链路是这样的：用户输入问题，首先进入意图识别模块，由它判断这个问题需要查数据库、查文档还是不需要查数据直接回答。如果需要查数据，就进入数据检索环节；检索结果交给可视化模块判断是否适合出图；最后所有输出汇总成一致的回复结构返回到前端。

这套架构里最容易被忽略但实际最影响体验的是"数据源管理模块"。没有它，检索 Agent 就是个瞎子——它不知道有哪些表、哪些字段、数据在哪个库。所以现在的主流做法是维护一份数据字典，包含表名、字段名、字段类型、字段注释、示例值，通过 Embedding 的方式灌入向量库。检索时先从向量库召回相关表结构，再把这些结构信息作为上下文拼接进 Text2SQL 的 Prompt 里。

2.2 为什么选 LangGraph 作为编排框架

我把 Langchain 和 LangGraph 的差异放一张表里，你就能直观理解选型逻辑：

这个项目明显属于后者。数据检索 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(model="gpt-4o-mini", temperature=0) 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(orient="records")}

这里有几个坑我必须提醒你：

第一，必须加 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(model="text-embedding-3-small") vectorstore = Chroma( persist_directory="./chroma_db", embedding_function=embeddings ) retriever = vectorstore.as_retriever( search_type="mmr", # 最大边际相关性，避免召回结果太相似 search_kwargs={"k": 6, "lambda_mult": 0.7} )

混合检索里一个容易被忽视的参数是召回条数。太多会稀释有效信息，太少可能漏掉关键内容。我做过的大多数项目里，k 取 4-8 是比较合理的区间。如果你用的是多路召回加融合排序，每路召回的 k 可以更小，但是路数可以多一些。

4. 可视化 Agent 的工程实现

4.1 从数据到图表类型的选择逻辑

可视化 Agent 最常见的失败模式是：用户明明要一个趋势图，它却搞了一个饼图。这可能不是模型能力问题，是你在 Prompt 里没有把图表类型的选择规则写清楚。

我总结了一套选择规则，可以直接写进系统提示词：

这套规则放进 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_ascii=False) 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 前端页面的布局与交互设计

页面布局上，我建议做成经典的"左侧对话流 + 右侧图表区"双栏结构。对话流里展示用户提问和系统回答，回答里包含文字摘要和对应的图表卡片；右侧区域放大展示最近一张图表，支持下载、全屏预览。

前端不要过度设计，这里功能优先级排序是：

1. 输入框可多行输入、支持回车发送
2. 对话气泡可区分用户和系统
3. 图表卡片随回答流插入
4. 每次新查询自动滚动到最新消息

如果项目里前端部分比较弱，也可以直接用 Plotly 或者 PyG2 生成 HTML 片段，然后通过 iframe 嵌进来。这样后端可以全包，前端开发成本基本为零，适合快速演示。

6. 核心 Agent Prompt 与工具链配置

6.1 各角色 Prompt 设计要点

下面我按角色把 Prompt 设计的核心要点拆一遍，每一条都是踩过坑之后总结出来的。

Planner Agent 的 Prompt 核心要素：

• 要求输出结构化 plan，每个子任务必须有明确的依赖关系
• 使用 few-shot 示例，至少给 3 个不同类型的问题样例
• 明确声明：做不到的事情直接说做不到，不要强行拆子任务
• 指定输出为 JSON 数组，字段包含 task_id、description、dependencies、expected_output

Query Agent 的 Prompt 核心要素：

• 明确数据字典的格式和使用方式，告诉模型优先使用注释字段来理解业务含义
• 给出一组"好 SQL 示例"和"坏 SQL 示例"
• 要求模型执行查询前先在内部确认表是否存在、字段是否匹配
• 输出时必须附带查询用的 SQL，供审计和调试

Visualize Agent 的 Prompt 核心要素：

• 明确图表类型选择的判断路径
• 强调字段必须来自数据结果集，不允许虚构字段
• 要求在配置生成后做一次检查，确认 X 轴字段不是数值型但被当成了连续轴，或者字段名拼写错误

Critic Agent 的 Prompt 核心要素：

• 审查 SQL 结果的数据类型是否符合预期
• 判断查询结果是否为空，并给出可能原因
• 检查图表配置的 data 字段是否与 review 数据一致
• 输出 pass/fail + 修改建议，反馈给对应的 Agent

6.2 Langchain 工具链封装：从函数到 Tool

Langchain 里创建工具非常直接，你可以用 @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(orient="records") @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_ascii=False)

工具描述是 Langchain 工具里最容易被忽略但最有价值的部分。好的工具描述应当做到：让模型读到描述时就能清楚知道"什么情况下调用这个工具"、"输入参数各代表什么"、"有什么注意事项"。另一个讲究是工具粒度。工具太粗，模型不知道该传什么参数；工具太细，工具列表过长影响推理速度和准确性。我个人的经验是一个 Agent 挂在 4-6 个工具以内，每个工具职责单一，描述不超过 100 字。

6.3 环境配置与依赖安装

这个项目你需要准备的环境：

• Python 3.10 或 3.11，3.9 以下跑不了新版 Langchain
• 一个可用的 LLM API，OpenAI 兼容接口即可
• 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( model="gpt-4o-mini", api_key="your-api-key", base_url="https://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_KEY=sk-xxxx OPENAI_API_BASE=https://api.openai.com/v1 DB_HOST=localhost DB_PORT=3306 DB_USER=root DB_PASSWORD=yourpassword DB_NAME=test_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表的结构，然后生成 SQL：

SELECT 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 优化编排。每一步的验证成本都低，出了问题也容易定位。这样走下来，你得到的不仅是一个能跑的系统，还会对每一层背后的原理有更深的体感。

本文还有配套的精品资源，点击获取