简介多智能体系统是当前AI应用工程化的重要方向它将复杂任务拆解为多个专业角色的协作流程让每个智能体只专注于自身职责从而显著提升输出稳定性与可控性。其核心原理是通过状态机和条件边来编排任务流使规划、检索、分析、审查等节点能够按逻辑顺序执行并在必要时进行循环修正。这种设计不仅解决了单一大模型提示词容易冲突的问题还使系统具备更强的可维护性与灵活性。在业务问答、报表生成、内部知识检索等场景中多智能体与RAG、可视化技术的结合尤为实用。本文基于一个真实项目完整拆解了使用LangChain与LangGraph构建的多智能体数据检索与可视化系统涵盖角色定义、工具封装、状态流转、图表生成及部署优化等关键环节为开发者提供了一套可参考的落地路径。 前阵子整理移动硬盘翻出一个归档工程名字就叫“基于Langchain多智能体数据检索与可视化系统.zip”。这个项目是我集中折腾Langchain那段时间的产物也是我认为Langchain这类框架最能出成果的组合方向把复杂的业务问答拆给多个智能体让专门的Agent负责数据检索让专门的Agent负责分析论证再让专门的角色出来挑毛病最终把结论用ECharts渲染到网页上。你如果已经写过几个基础Langchain Demo想在多智能体、数据检索、可视化这条链路上搭一个能演示、能扩展的真实系统这个项目的拆解思路应该能直接帮到你。整个系统解决的核心问题很直白用户在前端页面上输入一句业务问题比如“上季度华东区销售额为什么下滑”系统会把这个问题丢给一组分工明确的智能体有的负责拆解任务有的负责查SQL数据库有的负责查内部文档有的负责写分析报告还有一个专门负责抬杠找漏洞最后自动生成图表回传到页面。整个过程不需要用户关心数据在哪张表、图表怎么画。下面的拆解按四条线走工程结构、数据检索层、多智能体协作、可视化与部署。每个模块我都会把当时设计时的真实考虑、后来调过的参数、踩过的坑一起写完。1. 项目概览一个zip工程解决什么问题1.1 工程结构一拆模块职责先分清一个压缩包打开就是一套完整的前后端工程我第一次解压的时候目录是这样的langchain_multiagent_retrieval_viz/ ├── app.py # Flask 入口Web服务 ├── agents/ │ ├── planner.py # 任务规划智能体 │ ├── retriever.py # 数据检索智能体 │ ├── analyst.py # 分析报告智能体 │ ├── critic.py # 裁判/审查智能体 │ └── coordinator.py # 基于LangGraph的编排器 ├── tools/ │ ├── db_tool.py # SQL查询工具 │ ├── doc_tool.py # 文档检索工具 │ └── chart_tool.py # 图表配置生成工具 ├── visual/ │ ├── templates/ │ │ └── index.html │ └── static/ │ ├── echarts.min.js │ └── style.css ├── data/ │ ├── sales.db │ └── knowledge_docs/ ├── config.py └── requirements.txt当时刻意按智能体拆目录而不是按“登录模块、查询模块”那样拆是因为多智能体项目的核心维护点在角色身上。你想往系统里加一个角色新建一个py文件、注册到coordinator里就行不需要动业务代码。tools目录单独拎出来也很关键所有Agent共用的工具都放这里避免每个Agent都写一遍数据库连接逻辑。1.2 为什么这个场景必须上多智能体最早我用的是单个Agent一把梭把检索、分析、绘图全部塞进一个system prompt里结果特别不稳定。你要让这个Agent知道怎么查数据库又要让它会分析业务还要让它输出图表配置三件事挤在一个上下文里它会经常搞混。最典型的翻车是本来应该执行SQL查询它偏要先编一段分析结论或者让它输出JSON图表它非要带一段Markdown解释。这就像你让一个新人同时干产品经理、数据工程师、数据分析师三个岗位指令冲突时必然出问题。多智能体其实就是把“一个什么都会的程序员”拆成“需求分析师、数据工程师、业务分析师、代码审查员”。每个角色只负责一件事上下文更短提示词更纯粹输出稳定性明显提升。尤其是检索和生成这种能力取向完全不同的任务拆开之后效果是质的飞跃。而且拆开之后你还能单独对某个环节做优化——检索不准就换Embedding模型分析逻辑不对就只调analyst的prompt不会牵一发动全身。1.3 技术选型Langchain原生Agent还是LangGraph这是当时纠结最久的一个点。Langchain原生的AgentExecutor确实能用它是一个单Agent循环Agent思考、调用工具、观察结果、再思考直到产出最终答案。但一旦涉及多个Agent协作原生那套就不太够了因为你得自己写循环去控制角色A跑完再跑角色B还要处理各种条件分支比如裁判不满意要不要重跑分析Agent。写到最后要么是一堆callback嵌套要么是隐晦的while循环代码很快就变成屎山。LangGraph的定位完全不同它是给LLM应用用的状态机有Node节点、Edge边、State状态。你显式定义谁先谁后、什么情况下回退、什么情况下结束整个流程是可控的。我当时是这么取舍的对比项原生AgentExecutorLangGraph编程模型单Agent循环有向图 状态机多Agent编排需要手动写循环原生支持节点跳转条件分支有限条件边可控性强调试体验主要靠print日志每个节点都能检查状态适用场景单Agent单工具简单任务多Agent、需要状态流转的复杂任务注意LangGraph并不是替代Langchain它本身就是Langchain生态里的库底层继续用Langchain的LLM封装、Tool规范和Prompt模板。这套项目最终是LangGraph做编排骨架Langchain提供工具调用和模型接入能力两者配合使用。简单场景用Langchain原生Agent没问题但多智能体协作尤其是有“反馈-修正”这类循环需求时直接上LangGraph能少走很多弯路。2. 数据检索层实现让Agent学会查库和翻文档2.1 检索链路设计从一句问话到一份查询计划数据检索不是简单把用户问题扔给数据库执行就完事。用户问“上季度华东区销售额为什么下滑”这个问句本身不能直接翻译成SQL得先拆出查询条件时间范围是上季度地区是华东指标是销售额再决定要查哪张表、关联哪些维度。这就是planner智能体干的事。planner的输出是一份结构化的查询计划包含intent、search_terms、table、time_range这些字段。retriever拿到计划后判断该查SQL还是查文档还是两者都查。还有一种情况是用户问题本身模糊比如“最近业绩怎么样”这时候planner要识别出缺少时间范围向用户追问或者默认取最近一个月。这个追问逻辑我一开始没做结果用户输入模糊问题时系统经常查出一堆全量数据分析自然也是错的。这里有一个很重要的设计planner和retriever之间传递的一定是结构化数据不能是自然语言。自然语言在下游解析时大概率出问题。我当时用Pydantic定义了一个QueryPlan模型planner必须输出这个模型的实例retriever才能继续执行。2.2 工具封装给LLM写好说明书在Langchain里一个工具能否被Agent正确使用docstring写得好不好占一半以上。很多人觉得docstring是写给人看的但在Langchain里docstring就是LLM的“使用说明书”LLM靠它判断工具是干什么的、参数怎么传。我封装SQL查询工具时是这样写的from langchain_core.tools import tool import sqlite3 tool def query_sales_db(sql: str, limit: int 20) - str: Execute a SQL query against the sales database. The database has the following core tables: - sales_records: columns include region, product, sales_amount, order_date - regions: columns include region, manager, team_size Use precise SQL with WHERE clauses to filter by time and region. Always limit the result rows to avoid large responses. Return the result as plain text rows in column: value format. conn sqlite3.connect(data/sales.db) cur conn.cursor() safe_sql fSELECT * FROM ({sql.rstrip(;)}) LIMIT {limit} cur.execute(safe_sql) cols [desc[0] for desc in cur.description] rows cur.fetchall() conn.close() text \n.join([, .join(f{c}: {v} for c, v in zip(cols, row)) for row in rows]) return text or NO_RESULT这个docstring里必须包含两样东西一是表结构和关键字段名二是“尽量加WHERE条件、限制返回行数”这类使用规范。LLM是真会按你的docstring来写SQL的如果没告诉它有哪些表它会自己编一个不存在的表名。这个函数里我做了一次防注入处理把传入SQL的外层包一层子查询再统一LIMIT避免Agent一次查出上万行数据把上下文直接撑爆。文档检索工具用的就是标准RAG链路文档先切块用Embedding模型向量化存到FAISS索引。检索时根据query取top_k个相似块拼成上下文。这类工具docstring要写清楚“返回的是文档片段拼接文本用于回答背景知识类问题”不然Agent可能会尝试从文档片段里直接执行计算。2.3 把检索结果变成结构化数据检索Agent拿到原始查询结果后不能直接把文本丢给分析Agent。SQL返回的是票面记录比如一行一行的“region: 华东, sales_amount: 12000”分析Agent如果自己去数一共有多少行极容易数错LLM对数字的感知能力本来就不稳定。所以我在retriever和analyst之间加了一层结果整理逻辑检索Agent在返回结果之前会先调用一次LLM把原始数据转成聚合统计信息。比如原始SQL返回了100条订单记录LLM会基于这批数据生成“总销售额、环比变化、TOP3产品、各区域占比”这类摘要以JSON格式输出。这一步看起来多了一次LLM调用成本增加了但换来的是下游分析准确率大幅提升非常值得。结构化输出我用的是Langchain的PydanticOutputParser。注意这个parser不是百分百可靠LLM偶尔会输出残缺JSON。我的处理是try/except捕获解析失败然后把错误信息重新喂给LLM让它“根据错误信息修正后重新输出”最多重试两次。重试两次还失败就降级成正则提取再不行就把原始文本直接传给分析Agent。这个降级链路救了我好几次尤其是在切换模型版本的时候不同模型的JSON输出格式习惯差异挺大。3. 多智能体协作落地角色分工和LangGraph状态机3.1 角色分工与提示词工程多智能体系统里提示词工程的关键不是把每个角色的提示词写得多漂亮而是把每个角色的边界写清楚。边界越窄行为越可控。我当时四个角色的system prompt各有侧重planner的提示词核心是“不要尝试回答问题只输出查询计划”。它要避免一上来就给结论必须把用户问题拆成可执行的检索任务。analyst的提示词核心是“严格基于提供的数据分析不要编造数据”。所有结论必须能追溯到上一步检索返回的具体数字如果数据缺失直接说明缺失决不允许推测补全。critic的提示词核心是“你的任务是找问题”。它的存在就是为了挑错。每个提示词我还会加一句“如果用户问题不明确直接说明缺什么信息不要臆测”。这句话能省很多事情。很多Agent出错不是因为模型不行而是提示词里没有给它“承认不知道”的选项它只能硬着头皮编。3.2 正反博弈加裁判让结论经得起抬杠这算是这套系统里最有意思的一个机制。单纯让analyst分析数据它经常给出一版看起来合理、实际上逻辑跳步的结论。后来我加了正反博弈机制一个正方Agent负责生成结论一个反方Agent专门找茬最后由裁判Agent综合两边意见产出最终结论。这符合搜索引擎里常见的“红蓝对抗”思想让结论在批判中被修正。反方Agent的system prompt我写得很直接你是数据报告的反方审查者。你拿到一份分析结论后必须按以下方式质疑 1. 数据口径是否正确能否从原文数据中追溯到每一个数字 2. 是否存在因果颠倒或虚假相关 3. 是否遗漏了关键指标比如只看总额不看比率 4. 结论是否过度泛化样本是否支撑这个说法 5. 如果确实找不到问题必须输出无重大缺陷不允许为了挑错而编造问题。强调“无重大缺陷”这个兜底选项很重要不加这句反方Agent会为了完成“质疑”这个任务而硬挑毛病陷入为杠而杠的死循环。裁判Agent拿到正反两方的内容后输出最终结论同时附带图表配置。裁判的prompt里有一条我后面补的经验裁判可以不同意正反双方但不能引入检索结果之外的新数据。这一条保证了最终结论始终可溯源。3.3 LangGraph状态机让角色按流程跑起来LangGraph的核心用法是定义State、节点和边。State是一个TypedDict承载整个流程的中间数据。我这套系统的State差不多长这样from typing import TypedDict, List, Dict, Optional from langgraph.graph import StateGraph, END class AgentState(TypedDict): query: str plan: Optional[dict] retrieval_results: Optional[dict] report: Optional[str] critique: Optional[str] final_report: Optional[str] chart_config: Optional[dict] iteration: int needs_revision: bool节点函数接收State并返回一个dictLangGraph会把返回值更新到State里。plan_node负责调plannerretrieve_node负责调retrieveranalyze_node负责调analystcritique_node负责调criticfinalize_node负责汇总输出。关键是critique_node之后的条件边from langgraph.graph import StateGraph, END MAX_ITERATIONS 3 def should_continue(state: AgentState) - str: if state.get(needs_revision) and state.get(iteration, 0) MAX_ITERATIONS: return analyze return finalize graph StateGraph(AgentState) graph.add_node(plan, plan_node) graph.add_node(retrieve, retrieve_node) graph.add_node(analyze, analyze_node) graph.add_node(critique, critique_node) graph.add_node(finalize, finalize_node) graph.add_edge(plan, retrieve) graph.add_edge(retrieve, analyze) graph.add_edge(analyze, critique) graph.add_conditional_edges(critique, should_continue, { analyze: analyze, finalize: finalize }) graph.add_edge(finalize, END) app graph.compile()这个设计意味着如果反方Agent认为报告有问题系统会带着批判意见重新跑一遍analyst让analyst修正然后再交给critic重新审最多循环三次。iteration这个字段就是为了防止无限循环。实际跑下来80%的问题在第一轮交锋就能解决能跑到第三轮的情况已经很少。这个正向循环是单Agent结构无法优雅实现的也是LangGraph价值体现最明显的地方。4. 可视化系统实现从分析报告到ECharts图表4.1 为什么是Flask加ECharts而不是Streamlit开发可视化界面时我认真对比过Streamlit和Gradio它们确实开发速度快一个页面几分钟就能搭出来。但这套系统的定位不是Demo是要能对接企业内部数据的工具型产品。Streamlit适合分析型个人工具但做查询交互、权限控制、嵌入企业门户都别扭。Flask是最稳的Web框架想怎么控制都行前后端完全分离。ECharts则是我认为国内业务场景下图表库的最佳选择图表类型丰富、中文文档全、社区方案多渲染性能和交互性也够用。页面结构很简单顶部一个输入框用户输入业务问题并点击查询中间是分析报告区域展示最终结论下面是图表区渲染一到三个图表。关键一点是用户的问题不是一次性下单而是可以在页面上连续追问。比如第一次问“华东区销售额”得到结果后再问“细到产品品类”系统要把上一次查询上下文带入下一轮。这个逻辑我在后端用一个session级别的上下文对象来维护每次新查询会带上之前的检索摘要。4.2 图表配置的生成链路多智能体的最终产物不只是文本报告还有一个图表配置JSON。裁判Agent输出的final_report是自然语言不能直接拿去绘图。我让finalize节点调用chart_tool把final_report里的核心数字和维度提取出来转成图表配置。一个标准的bar chart配置长这样{ chart_type: bar, title: 2025年第一季度各业务线营收, x_axis: [华东, 华北, 华南], series: [ {name: 营收, data: [1200, 980, 1450]} ] }前端拿到这个JSON后调用echarts.init和setOption两步就能渲染。这个链路最关键的地方在chart_tool里不能让LLM完全自由发挥。我定义了一个ChartSchema用Pydantic强行约束chart_type必须是bar、line、pie中的一种x_axis不能为空series里data长度必须与x_axis一致。校验不过就让LLM重新生成最多重试两次。这套约束下来前端从来没有因为配置问题白屏过。4.3 查询接口的缓存与并发优化多智能体查询链路比较长用户每次点查询要串行跑plan、retrieve、analyze、critique、finalize五步逐次调用LLM耗时少则十几秒多则一分钟。前几版没有缓存同一句话敲两遍就白花两份API钱。后来我在后端加了一层内存缓存缓存key由用户问题加上下文摘要拼成值为最终报告和图表JSON。命中后直接返回不再调用LLM。对于耗时较长的问题轮询比纯等待体验好很多。我采用的方式是Flask收到查询请求后启动一个后台线程执行多智能体任务立刻返回一个task_id。前端每两秒GET一次/query/status拿到SUCCESS或者FAILED状态后再取结果。线程池我直接用concurrent.futures.ThreadPoolExecutor最大线程数设为4任务队列撑满后让用户稍后再试。这套轻量方案没有引入Redis和Celery对于一个内部工具项目来说已经够稳定了。5. 部署与性能优化让系统真正跑起来5.1 依赖锁定与环境部署Langchain生态的迭代速度快到离谱几乎每个月都有破坏性变更。这个项目里我踩过一个经典版本坑requirements.txt里写着langchain一个版本跑起来报deprecation warning改完代码后另一篇教程里又用了新API最后整个项目处于“老教程跑不通、新代码不敢改”的尴尬状态。所以如果你打算照着这个项目做第一件事就是把依赖锁死。我最终的requirements.txt是长期打磨出来的版本langchain0.3.7 langchain-openai0.2.8 langgraph0.2.44 langchain-community0.3.7 faiss-cpu1.8.0 flask3.0.3 pydantic2.9.2 gunicorn23.0.0部署时我用gunicorn起Flask服务配置4个worker。注意如果你的检索链路里用了长连接数据库资源多worker模式下会有连接数翻倍的问题。我在工具函数里用的是sqlite这个基本无压力如果是MySQL这类服务器建议每个工具都显式管理连接用完即关。环境变量方面OpenAI的API key、Embedding模型名称、数据库路径全部放进config.py统一管理用os.getenv读取。另外一个部署建议是Embedding模型尽量用开源的本地模型比如BGE系列检索质量不输商业API太远而且不必担心网络调用失败。文档向量化是离线批量做的生成FAISS索引文件程序启动时加载到内存减少在线计算开销。5.2 成本、性能优化清单我把实际运行中最有效的几个调优点整理出来了都是直接可见的收益第一向量索引优化。FAISS默认的flat索引在小数据集上没问题但文档量到几十万条后检索变慢我把索引换成IVF索引nlist根据数据量设为数据开根号约100nprobe设为10检索返回时间快了几乎一个数量级精度损失可以忽略。第二SQL查询限制返回行数。我在query_sales_db工具里统一加了LIMIT默认20行。如果没有LIMITAgent可能写出一条返回全表数据的SQL不仅上下文爆炸数据库也扛不住。AI生成的SQL质量通常不够稳定工具层必须兜底。第三LLM调用缓存。除了最终报告缓存我在检索整理那一步也加了缓存。相同的数据摘要不会重复调用LLM聚合能在多轮追问场景下省下不少成本。对于tokengpt成本敏感的项目用便宜的小模型做planner的意图识别用强模型做analyst和critic这个分层调用组合实测下来效果很好。6. 常见问题与排查实录6.1 问题速查表多智能体系统的坑比单Agent系统多。以下这些问题我全部实际遇到过整理出来可以直接对照定位现象可能原因解决方案Agent一直循环不结束工具返回格式不匹配Agent不断重试增加max_iterations在工具函数严格按文档格式返回上下文爆掉检索结果行数太多或文档块过大限制SQL的LIMIT调小向量检索的top_k裁判永远在批评裁判提示词没有兜底语义增加“无重大缺陷”输出选项前端图表渲染为空图表JSON配置缺字段用Pydantic定义ChartSchema强校验不合格重新生成时间字段显示为时间戳SQL返回datetime未转字符串在工具层统一把字段转为yyyy-MM-dd格式多轮追问丢失上下文session上下文未传递前端请求带上session_id后端维护上下文对象6.2 三个印象最深的排查案例第一个案例是Agent死循环。自定义工具的返回里带了多余的说明文字比如“查询成功共找到5条记录其中华东区域数据如下”LLM把说明文字当成真实数据结构反复尝试重新查询就是不往下走。排查半天后发现是工具返回里多了“查询成功”这句话。这之后我立了一个规矩工具返回只能用纯数据文本任何提示文字都不要加。让LLM判断数据质量是它自己的事工具只负责把结果搬过来越简单越好。第二个案例是检索结果和图表数值对不上。裁判Agent给出的报告里说“华东区营收1200万”ECharts柱状图华东柱子的数值却是980。查下来是chart_tool生成图表配置时模型在重述数据的过程中把数字搞错了。这个问题的根源是不该让生成型模型二次转述具体数字。修正方式是不再让chart_tool单独生成数值而是由retrieve节点的原始结构化数据里直接读取数值chart_tool只负责决定图表类型和字段映射。从那之后图表里的数字基本不会错了。第三个案例是Flask多线程下的状态串扰。当两个用户同时提问我最初把AgentState定义成模块级全局变量结果请求A和请求B的状态互相污染明明用户A在问华东区生成结果里却出现了华北区的数据。这是因为两个请求共享了同一个State对象。修复方式是每个请求在进入LangGraph编译对象之前显式创建一个独立的状态字典。LangGraph的图编译实例本身可以复用但每次执行时初始state必须新建。7. 这套系统后续还能怎么扩展如果要我给这个项目规划下一步我会优先做两个方向。第一个方向是给planner加数据字典工具先把数据库所有的表结构、字段注释、枚举值做成一个工具注册进去让检索Agent自己根据用户问题来定位该查哪张表。第二个方向是把上下文记忆做成真正长期化目前session级别的上下文只在一个查询会话内有效如果用户隔一天再次问同一个业务线的问题系统不记得之前的分析结论如果接上向量数据库的记忆存储多轮业务追踪会顺畅很多。归根结底多智能体的代码写起来并不复杂复杂的是把每个角色的边界、状态流转和工具协议弄清楚。我搭这套系统最大的体会是别为了多智能体而多智能体先想清楚你的任务链路里哪里需要分角色、哪里需要反馈循环再决定拆几个Agent。踩过几次坑之后你会发现真正让系统稳定跑起来的不是模型多聪明而是工程边界划得多清楚。这套拆解里的方案和坑希望对打算做同类项目的你有参考价值。本文还有配套的精品资源点击获取