企业官网建设流程全解析

一、实现流程介绍

在本专栏的前面文章中，介绍过基于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，以柱状图的方式呈现给我

本次实验所依赖的版本如下所示：

mcp==1.9.2openai==1.75.0langchain==0.3.25 langchain-openai==0.3.18langgraph==0.4.7pymysql==1.0.3

二、搭建 数据查询 MCP Server

首先创建数据表，并导入上述数据：

CREATETABLE`us_covid19_counties`(`date`varchar(255)CHARACTERSETutf8mb4COLLATEutf8mb4_0900_ai_ciDEFAULTNULLCOMMENT'日期',`county`varchar(255)CHARACTERSETutf8mb4COLLATEutf8mb4_0900_ai_ciDEFAULTNULLCOMMENT'县',`state`varchar(255)CHARACTERSETutf8mb4COLLATEutf8mb4_0900_ai_ciDEFAULTNULLCOMMENT'州',`fips`varchar(255)CHARACTERSETutf8mb4COLLATEutf8mb4_0900_ai_ciDEFAULTNULLCOMMENT'县编码code',`cases`intDEFAULTNULLCOMMENT'累计确诊病例',`deaths`intDEFAULTNULLCOMMENT'累计死亡病例')ENGINE=InnoDBDEFAULTCHARSET=utf8mb4COLLATE=utf8mb4_0900_ai_ciCOMMENT='COVID-19 案例';

实现数据库操作，这里仅仅做了数据库交互，实际使用时应考虑很多性能细节的优化：

utils_db.py

importpymysqldefget_conn():returnpymysql.connect(host="127.0.0.1",port=3306,database="test3",user="root",password="root",autocommit=True)defquery(sql):conn=get_conn()cursor=conn.cursor()cursor.execute(sql)columns=[column[0]forcolumnincursor.description]res=list()forrowincursor.fetchall():res.append(dict(zip(columns,row)))cursor.close()conn.close()returnres

创建数据查询MCP Server，这里实现了两个MCP Tools， 分别是获取表Schema和 执行SQL两个操作：

importjsonfrommcp.server.fastmcpimportFastMCPimportutils_db mcp=FastMCP("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:schemas=utils_db.query(schema_sql.format(table=table))schemas=", ".join([f"{s['COLUMN_NAME']}{s['DATA_TYPE']}COMMENT{s['COLUMN_COMMENT']}"forsinschemas])table_schema.append(f"TABLE `{table}` ({schemas})")return"\n\n".join(table_schema)@mcp.tool()defrun_sql(sql:str)->str:"""执行MySQL SQL语句查询数据，一次仅能执行一句SQL！"""try:returnstr(utils_db.query(sql))exceptExceptionase:returnf"执行SQL错误：{str(e)},请修正后重新发起。"if__name__=="__main__":mcp.settings.port=6030mcp.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():client=MultiServerMCPClient({"chart":{"url":"https://mcp.api-inference.modelscope.net/530c948576c248/sse","transport":"sse"},"covid19-query":{"url":"http://127.0.0.1:6030/sse","transport":"sse"}})tools=awaitclient.get_tools()checkpointer=InMemorySaver()agent=create_react_agent("openai:gpt-4.1",tools,checkpointer=checkpointer,prompt="You are an AI assistant that helps people find information.")config={"configurable":{"thread_id":"1"}}whileTrue:question=input("请输入：")ifnotquestion:continueifquestion=="q":breakasyncforchunkinagent.astream({"messages":[{"role":"user","content":question}]},stream_mode="updates",config=config):if"agent"inchunk:content=chunk["agent"]["messages"][0].content tool_calls=chunk["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,f"LLM:{content}")elif"tools"inchunk:content=chunk["tools"]["messages"][0].content name=chunk["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 app=FastAPI(title="OpenAI Compatible Chat API")api_key="sk-da4b6cb4a41e4cascascasc9508deb556942"agent=NoneclassChatCompletionRequest(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:str="chat.completion"created:intmodel:strchoices:List[ChatCompletionChoice]usage:Dict[str,int]classChatCompletionChunk(BaseModel):id:strobject:str="chat.completion.chunk"created:intmodel:strchoices:List[Dict[str,Any]]asyncdefcreate_agent():'''创建MCP Agent'''client=MultiServerMCPClient({"chart":{"url":"https://mcp.api-inference.modelscope.net/530c948576c248/sse","transport":"sse"},"covid19-query":{"url":"http://127.0.0.1:6030/sse","transport":"sse"}})tools=awaitclient.get_tools()returncreate_react_agent("openai:gpt-4.1",,tools,prompt="You are an AI assistant that helps people find information.\n针对用户的问题，你的执行流程是1.获取表结构。2.生成SQL查询数据。3.生成可视化图表。4.总结结果回复用户")asyncdefinvoke_agent(messages:[]):globalagentifnotagent:agent=awaitcreate_agent()asyncforchunkinagent.astream({"messages":messages},stream_mode="updates"):if"agent"inchunk:content=chunk["agent"]["messages"][0].content tool_calls=chunk["agent"]["messages"][0].tool_callsiftool_calls:fortoolintool_calls:yieldf"> ```Call MCP Server:{tool['name']}```\n\n"else:yieldcontentasyncdefhandle_stream_response(messages:[],model:str,request_id:str)->Generator[str,None,None]:# 执行 agentasyncformsgininvoke_agent(messages):chunk_data=ChatCompletionChunk(id=request_id,created=int(time.time()),model=model,choices=[{"index":0,"delta":{"content":msg},"finish_reason":None}])yieldf"data:{chunk_data.model_dump_json()}\n\n"final_chunk=ChatCompletionChunk(id=request_id,created=int(time.time()),model=model,choices=[{"index":0,"delta":{},"finish_reason":"stop"}])yieldf"data:{final_chunk.model_dump_json()}\n\n"yield"data: [DONE]\n\n"asyncdefverify_auth(authorization:Optional[str]=Header(None))->bool:'''验证token'''ifnotauthorization:returnFalseifauthorization.startswith("Bearer "):token=authorization[7:]else:token=authorizationreturntoken==api_key@app.post("/v1/chat/completions")asyncdefchat_completions(request:ChatCompletionRequest,auth_result:bool=Depends(verify_auth)):# 检查身份验证结果ifnotauth_result:raiseHTTPException(status_code=401,detail={"error":{"message":"Invalid authentication credentials","type":"invalid_request_error","param":None,"code":"invalid_api_key"}},headers={"WWW-Authenticate":"Bearer"})## 暂不支持非流式返回ifnotrequest.stream:raiseHTTPException(status_code=400,detail={"error":{"message":"Streaming responses are not implemented in this mock API","type":"invalid_request_error","param":"stream","code":"invalid_parameter"}})try:# 触发 agent 并流式返回request_id=f"chatcmpl-{uuid.uuid4().hex[:8]}"returnStreamingResponse(handle_stream_response(request.messages,request.model,request_id),media_type="text/plain",headers={"Cache-Control":"no-cache","Connection":"keep-alive","Content-Type":"text/event-stream"})exceptExceptionase:raiseHTTPException(status_code=500,detail=str(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,host="0.0.0.0",port=8000)

启动服务：

五、使用 OpenAI 协议客户端进行测试

这里以Cherry Studio为例，你也可以换成OpenWebUI或者其他支持OpenAI协议的客户端：

配置连接：

对话测试：

提问：统计的确诊数量前十名的州，以折线图的方式展示

提问：找出确诊人数前五名的州，按饼图的方式绘制给我

提问：根据每个县的确诊人数取Top10，以柱状图的方式呈现给我

提问：统计确诊数量最多的州占整体的百分比，以液体展示