手把手教你用LangChain + MCP 构建带可视化图表功能的ChatBI智能体，全程干货，草履虫也能学会！！

一、实现流程介绍在本专栏的前面文章中介绍过基于LangChain MCP和SpringAI MCP构建ChatBI智能体但未对数据进行可视化图表的分析而可视化图表是数据分析的重要组成部分它能够将复杂的数据转化为直观易懂的图形表示。因此本篇文章基于LangChain MCP实验构建ChatBI 可视化图表综合的智能体。使用数据说明其中分析数据采用数据采用COVID-19测试案例包括美国2021-01-28号各个县county的新冠疫情累计案例信息包括确诊病例和死亡病例数据格式如下所示date日期,county县,state州,fips县编码code,cases累计确诊病例,deaths累计死亡病例2021-01-28,Pike,Alabama,01109,2704,352021-01-28,Randolph,Alabama,01111,1505,372021-01-28,Russell,Alabama,01113,3675,162021-01-28,Shelby,Alabama,01117,19878,1412021-01-28,St.Clair,Alabama,01115,8047,1472021-01-28,Sumter,Alabama,01119,925,282021-01-28,Talladega,Alabama,01121,6711,1142021-01-28,Tallapoosa,Alabama,01123,3258,1122021-01-28,Tuscaloosa,Alabama,01125,22083,2832021-01-28,Walker,Alabama,01127,6105,1852021-01-28,Washington,Alabama,01129,1454,27下载地址如下https://github.com/BIXUECHAO/covid-19-tes-data/blob/main/us-covid19-counties.csv整体实现过程如下所示其中可视化图表能力采用ModelScope上公开的MCP-Server地址为https://modelscope.cn/mcp/servers/antvis/mcp-server-chart连接方式{mcpServers:{mcp-server-chart:{type:sse,url:https://mcp.api-inference.modelscope.net/530c948576c248/sse}}}Covid19 Query MCP Server则是通过FastMCP来自定义实现。大模型端采用OpenAI的GPT-4.1模型你也可以换成其他支持工具调通的模型。实验最终实现的效果如下所示提问找出确诊人数前五名的州按饼图的方式绘制给我提问根据每个县的确诊人数取Top10以柱状图的方式呈现给我本次实验所依赖的版本如下所示mcp1.9.2openai1.75.0langchain0.3.25langchain-openai0.3.18langgraph0.4.7pymysql1.0.3二、搭建 数据查询 MCP Server首先创建数据表并导入上述数据CREATE TABLE us_covid19_counties(date varchar(255)CHARACTER SET utf8mb4 COLLATE utf8mb4_0900_ai_ci DEFAULT NULL COMMENT日期,county varchar(255)CHARACTER SET utf8mb4 COLLATE utf8mb4_0900_ai_ci DEFAULT NULL COMMENT县,state varchar(255)CHARACTER SET utf8mb4 COLLATE utf8mb4_0900_ai_ci DEFAULT NULL COMMENT州,fips varchar(255)CHARACTER SET utf8mb4 COLLATE utf8mb4_0900_ai_ci DEFAULT NULL COMMENT县编码code,casesintDEFAULT NULL COMMENT累计确诊病例,deathsintDEFAULT NULL COMMENT累计死亡病例)ENGINEInnoDB DEFAULT CHARSETutf8mb4 COLLATEutf8mb4_0900_ai_ci COMMENTCOVID-19 案例;实现数据库操作这里仅仅做了数据库交互实际使用时应考虑很多性能细节的优化utils_db.pyimportpymysql defget_conn():returnpymysql.connect(host127.0.0.1,port3306,databasetest3,userroot,passwordroot,autocommitTrue)defquery(sql):connget_conn()cursorconn.cursor()cursor.execute(sql)columns[column[0]forcolumnincursor.description]reslist()forrowincursor.fetchall():res.append(dict(zip(columns,row)))cursor.close()conn.close()returnres创建数据查询MCP Server这里实现了两个MCP Tools 分别是获取表Schema和 执行SQL两个操作importjsonfrommcp.server.fastmcpimportFastMCPimportutils_db mcpFastMCP(COVID-19-DB-Query-Mcp-Server)schema_sql SELECT COLUMN_NAME, DATA_TYPE, COLUMN_COMMENT FROM INFORMATION_SCHEMA.COLUMNS WHERE TABLE_SCHEMA test3 AND TABLE_NAME {table} effective_tables[us_covid19_counties]mcp.tool()defget_table_schema()-str:获取当前有效表的 MySQL Schema, 生成SQL之前你应该使用该工具获取完整的表结构table_schema[]fortableineffective_tables:schemasutils_db.query(schema_sql.format(tabletable))schemas, .join([f{s[COLUMN_NAME]}{s[DATA_TYPE]}COMMENT{s[COLUMN_COMMENT]}forsinschemas])table_schema.append(fTABLE {table} ({schemas}))return\n\n.join(table_schema)mcp.tool()defrun_sql(sql:str)-str:执行MySQL SQL语句查询数据一次仅能执行一句SQLtry:returnstr(utils_db.query(sql))exceptExceptionase:returnf执行SQL错误{str(e)},请修正后重新发起。if__name____main__:mcp.settings.port6030mcp.run(sse)启动MCP Server三、LangChain MCP Client 智能体构建LangChain官方关于MCP的集成介绍文档如下https://langchain-ai.github.io/langgraph/agents/mcp/关于MCP的连接通过MultiServerMCPClient抽取出所有的MCP Tools然后交给create_react_agent构建智能体。实现过程和本地控制台测试逻辑如下importos,config os.environ[OPENAI_BASE_URL]https://api.openai.com/v1## 支持openai协议的 API BASEos.environ[OPENAI_API_KEY]sk-xxxxx## 你的API KEYfromlanggraph.prebuiltimportcreate_react_agentfromlangchain_mcp_adapters.clientimportMultiServerMCPClientfromcoloramaimportFore,Styleimportasynciofromlanggraph.checkpoint.memoryimportInMemorySaver asyncdefmain():clientMultiServerMCPClient({chart:{url:https://mcp.api-inference.modelscope.net/530c948576c248/sse,transport:sse},covid19-query:{url:http://127.0.0.1:6030/sse,transport:sse}})toolsawaitclient.get_tools()checkpointerInMemorySaver()agentcreate_react_agent(openai:gpt-4.1,tools,checkpointercheckpointer,promptYou are an AI assistant that helps people find information.)config{configurable:{thread_id:1}}whileTrue:questioninput(请输入)ifnot question:continueifquestionq:breakasyncfor chunkinagent.astream({messages:[{role:user,content:question}]},stream_modeupdates,configconfig):ifagentinchunk:contentchunk[agent][messages][0].content tool_callschunk[agent][messages][0].tool_callsiftool_calls:fortoolintool_calls:print(Fore.YELLOW,Style.BRIGHT,f Call MCP Server:{tool[name]}, args:{tool[args]})else:print(Fore.BLACK,Style.BRIGHT,fLLM:{content})eliftoolsinchunk:contentchunk[tools][messages][0].content namechunk[tools][messages][0].nameprint(Fore.GREEN,Style.BRIGHT,f{name}:{content})if__name____main__:asyncio.run(main())运行测试提问统计的确诊数量前十名的州以折线图的方式展示四、封装为 OpenAI 协议接口上述已经通过控制台实现了完整的Agent交互过程下面继续改造逻辑使用FastAPI封装为OpenAI协议接口便于在客户端使用。注意这里主要为实现功能api_key写死在程序中为sk-da4b6cb4a41e4cascascasc9508deb556942随机生成的 后续使用客户端连接时需要填写该api_key。importos os.environ[OPENAI_BASE_URL]https://api.openai.com/v1## 支持openai协议的 API BASEos.environ[OPENAI_API_KEY]sk-xxxxx## 你的API KEYfromfastapiimportFastAPI,HTTPException,Header,Dependsfromfastapi.responsesimportStreamingResponsefrompydanticimportBaseModelfromtyping importList,Dict,Any,Optional,Generatorfromlanggraph.prebuiltimportcreate_react_agentfromlangchain_mcp_adapters.clientimportMultiServerMCPClientfromdatetimeimportdatetimeimporttime,uuid appFastAPI(titleOpenAI Compatible Chat API)api_keysk-da4b6cb4a41e4cascascasc9508deb556942agentNoneclassChatCompletionRequest(BaseModel):model:strmessages:List[Dict[str,Any]]temperature:Optional[float]1.0max_tokens:Optional[int]Nonestream:Optional[bool]FalseclassChatCompletionChoice(BaseModel):index:intmessage:List[Dict[str,Any]]finish_reason:strclassChatCompletionResponse(BaseModel):id:strobject:strchat.completioncreated:intmodel:strchoices:List[ChatCompletionChoice]usage:Dict[str,int]classChatCompletionChunk(BaseModel):id:strobject:strchat.completion.chunkcreated:intmodel:strchoices:List[Dict[str,Any]]asyncdefcreate_agent():创建MCP AgentclientMultiServerMCPClient({chart:{url:https://mcp.api-inference.modelscope.net/530c948576c248/sse,transport:sse},covid19-query:{url:http://127.0.0.1:6030/sse,transport:sse}})toolsawaitclient.get_tools()returncreate_react_agent(openai:gpt-4.1,,tools,promptYou are an AI assistant that helps people find information.\n针对用户的问题你的执行流程是1.获取表结构。2.生成SQL查询数据。3.生成可视化图表。4.总结结果回复用户)asyncdefinvoke_agent(messages:[]):globalagent ifnot agent:agentawaitcreate_agent()asyncfor chunkinagent.astream({messages:messages},stream_modeupdates):ifagentinchunk:contentchunk[agent][messages][0].content tool_callschunk[agent][messages][0].tool_callsiftool_calls:fortoolintool_calls:yieldf Call MCP Server:{tool[name]}\n\nelse:yieldcontent asyncdefhandle_stream_response(messages:[],model:str,request_id:str)-Generator[str,None,None]:# 执行 agentasyncfor msgininvoke_agent(messages):chunk_dataChatCompletionChunk(idrequest_id,createdint(time.time()),modelmodel,choices[{index:0,delta:{content:msg},finish_reason:None}])yieldfdata:{chunk_data.model_dump_json()}\n\nfinal_chunkChatCompletionChunk(idrequest_id,createdint(time.time()),modelmodel,choices[{index:0,delta:{},finish_reason:stop}])yieldfdata:{final_chunk.model_dump_json()}\n\nyielddata: [DONE]\n\nasyncdefverify_auth(authorization:Optional[str]Header(None))-bool:验证tokenifnot authorization:returnFalseifauthorization.startswith(Bearer ):tokenauthorization[7:]else:tokenauthorizationreturntokenapi_keyapp.post(/v1/chat/completions)asyncdefchat_completions(request:ChatCompletionRequest,auth_result:boolDepends(verify_auth)):# 检查身份验证结果ifnot auth_result:raiseHTTPException(status_code401,detail{error:{message:Invalid authentication credentials,type:invalid_request_error,param:None,code:invalid_api_key}},headers{WWW-Authenticate:Bearer})## 暂不支持非流式返回ifnot request.stream:raiseHTTPException(status_code400,detail{error:{message:Streaming responses are not implemented in this mock API,type:invalid_request_error,param:stream,code:invalid_parameter}})try:# 触发 agent 并流式返回request_idfchatcmpl-{uuid.uuid4().hex[:8]}returnStreamingResponse(handle_stream_response(request.messages,request.model,request_id),media_typetext/plain,headers{Cache-Control:no-cache,Connection:keep-alive,Content-Type:text/event-stream})exceptExceptionase:raiseHTTPException(status_code500,detailstr(e))app.get(/v1/models)asyncdeflist_models():列出可用模型return{object:list,data:[{id:agent_model,object:model,created:int(time.time()),owned_by:agent}]}app.get(/health)asyncdefhealth_check():健康检查return{status:healthy,timestamp:datetime.now().isoformat()}if__name____main__:importuvicorn uvicorn.run(app,host0.0.0.0,port8000)启动服务五、使用 OpenAI 协议客户端进行测试这里以Cherry Studio为例你也可以换成OpenWebUI或者其他支持OpenAI协议的客户端配置连接对话测试提问统计的确诊数量前十名的州以折线图的方式展示提问找出确诊人数前五名的州按饼图的方式绘制给我提问根据每个县的确诊人数取Top10以柱状图的方式呈现给我提问统计确诊数量最多的州占整体的百分比以液体展示最后为什么要学AI大模型当下⼈⼯智能市场迎来了爆发期并逐渐进⼊以⼈⼯通⽤智能AGI为主导的新时代。企业纷纷官宣“ AI ”战略为新兴技术⼈才创造丰富的就业机会⼈才缺⼝将达 400 万DeepSeek问世以来生成式AI和大模型技术爆发式增长让很多岗位重新成了炙手可热的新星岗位薪资远超很多后端岗位在程序员中稳居前列。与此同时AI与各行各业深度融合飞速发展成为炙手可热的新风口企业非常需要了解AI、懂AI、会用AI的员工纷纷开出高薪招聘AI大模型相关岗位。最近很多程序员朋友都已经学习或者准备学习 AI 大模型后台也经常会有小伙伴咨询学习路线和学习资料我特别拜托北京清华大学学士和美国加州理工学院博士学位的鲁为民老师给大家这里给大家准备了一份涵盖了AI大模型入门学习思维导图、精品AI大模型学习书籍手册、视频教程、实战学习等录播视频全系列的学习资料这些学习资料不仅深入浅出而且非常实用让大家系统而高效地掌握AI大模型的各个知识点。这份完整版的大模型 AI 学习资料已经上传CSDN朋友们如果需要可以微信扫描下方CSDN官方认证二维码免费领取【保证100%免费】AI大模型系统学习路线在面对AI大模型开发领域的复杂与深入精准学习显得尤为重要。一份系统的技术路线图不仅能够帮助开发者清晰地了解从入门到精通所需掌握的知识点还能提供一条高效、有序的学习路径。但知道是一回事做又是另一回事初学者最常遇到的问题主要是理论知识缺乏、资源和工具的限制、模型理解和调试的复杂性在这基础上找到高质量的学习资源不浪费时间、不走弯路又是重中之重。AI大模型入门到实战的视频教程项目包看视频学习是一种高效、直观、灵活且富有吸引力的学习方式可以更直观地展示过程能有效提升学习兴趣和理解力是现在获取知识的重要途径光学理论是没用的要学会跟着一起敲要动手实操才能将自己的所学运用到实际当中去这时候可以搞点实战案例来学习。海量AI大模型必读的经典书籍PDF阅读AI大模型经典书籍可以帮助读者提高技术水平开拓视野掌握核心技术提高解决问题的能力同时也可以借鉴他人的经验。对于想要深入学习AI大模型开发的读者来说阅读经典书籍是非常有必要的。600AI大模型报告实时更新这套包含640份报告的合集涵盖了AI大模型的理论研究、技术实现、行业应用等多个方面。无论您是科研人员、工程师还是对AI大模型感兴趣的爱好者这套报告合集都将为您提供宝贵的信息和启示。AI大模型面试真题答案解析我们学习AI大模型必然是想找到高薪的工作下面这些面试题都是总结当前最新、最热、最高频的面试题并且每道题都有详细的答案面试前刷完这套面试题资料小小offer不在话下这份完整版的大模型 AI 学习资料已经上传CSDN朋友们如果需要可以微信扫描下方CSDN官方认证二维码免费领取【保证100%免费】