一、实现流程介绍在本专栏的前面文章中介绍过基于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.25 langchain-openai0.3.18langgraph0.4.7pymysql1.0.3二、搭建 数据查询 MCP Server首先创建数据表并导入上述数据CREATETABLEus_covid19_counties(datevarchar(255)CHARACTERSETutf8mb4COLLATEutf8mb4_0900_ai_ciDEFAULTNULLCOMMENT日期,countyvarchar(255)CHARACTERSETutf8mb4COLLATEutf8mb4_0900_ai_ciDEFAULTNULLCOMMENT县,statevarchar(255)CHARACTERSETutf8mb4COLLATEutf8mb4_0900_ai_ciDEFAULTNULLCOMMENT州,fipsvarchar(255)CHARACTERSETutf8mb4COLLATEutf8mb4_0900_ai_ciDEFAULTNULLCOMMENT县编码code,casesintDEFAULTNULLCOMMENT累计确诊病例,deathsintDEFAULTNULLCOMMENT累计死亡病例)ENGINEInnoDBDEFAULTCHARSETutf8mb4COLLATEutf8mb4_0900_ai_ciCOMMENTCOVID-19 案例;实现数据库操作这里仅仅做了数据库交互实际使用时应考虑很多性能细节的优化utils_db.pyimportpymysqldefget_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.memoryimportInMemorySaverasyncdefmain():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(请输入)ifnotquestion:continueifquestionq:breakasyncforchunkinagent.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.responsesimportStreamingResponsefrompydanticimportBaseModelfromtypingimportList,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:[]):globalagentifnotagent:agentawaitcreate_agent()asyncforchunkinagent.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:yieldcontentasyncdefhandle_stream_response(messages:[],model:str,request_id:str)-Generator[str,None,None]:# 执行 agentasyncformsgininvoke_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:验证tokenifnotauthorization:returnFalseifauthorization.startswith(Bearer ):tokenauthorization[7:]else:tokenauthorizationreturntokenapi_keyapp.post(/v1/chat/completions)asyncdefchat_completions(request:ChatCompletionRequest,auth_result:boolDepends(verify_auth)):# 检查身份验证结果ifnotauth_result:raiseHTTPException(status_code401,detail{error:{message:Invalid authentication credentials,type:invalid_request_error,param:None,code:invalid_api_key}},headers{WWW-Authenticate:Bearer})## 暂不支持非流式返回ifnotrequest.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以柱状图的方式呈现给我提问统计确诊数量最多的州占整体的百分比以液体展示
