一、私有化智能体应用在本专栏的前面文章基于Spring AI MCP实现了本地ChatBI问答应用本文还是依据该场景采用LangChain vLLM Qwen3-32B MCP技术栈构建该流程整体过程如下图所示实现效果如下所示关于MySQL表结构的创建可以参考下面这篇文章Spring AI MCP Server Cline 快速搭建一个数据库 ChatBi 助手实验所使用依赖的版本如下torch2.6.0transformers4.51.3modelscope1.23.1vllm0.8.4mcp1.9.2openai1.75.0langchain0.3.25 langchain-openai0.3.18langgraph0.4.7pymysql1.0.3二、vLLM 部署 Qwen3-32B使用modelscope下载Qwen3-32B模型到本地modelscope download--modelQwen/Qwen3-32B--local_dirQwen3-32BvLLM读取模型启动API服务。exportCUDA_VISIBLE_DEVICES0,1vllm serveQwen3-32B\--host0.0.0.0\--port8060\--dtypebfloat16\--tensor-parallel-size2\--cpu-offload-gb0\--gpu-memory-utilization0.8\--max-model-len8126\--api-key token-abc123\--enable-prefix-caching\--enable-reasoning\--reasoning-parser deepseek_r1\--enable-auto-tool-choice\--tool-call-parser hermes\--trust-remote-code关键参数说明export CUDA_VISIBLE_DEVICES0,1指定所使用的GPU。dtype: 数据类型其中bfloat1616位浮点数适合NVIDIA A100等设备。tensor-parallel-sizeTensor并行的数量当多GPU分布式推理时使用建议和GPU的数量一致。cpu-offload-gb允许将部分模型权重或中间结果卸载到CPU的内存中单位为GB。模拟GPU内存扩展如果部署的模型大于了显存大小可以设置该参数但是推理速度会大大下降。gpu-memory-utilization设置GPU内存利用率的上限。max-model-len允许模型最大处理的Token数该参数越大占用显存越大。enable-prefix-caching启用前缀缓存减少重复计算。enable-reasoning启用思考推理能力。reasoning-parser deepseek_r1指定推理解析器。enable-auto-tool-choice启用 function call 模式。tool-call-parser hermes设置 function call 的解析器。显存占用情况如果启动显存不足可适当调整gpu-memory-utilization和max-model-len参数或通过cpu-offload-gb将部分模型权重卸载到内存中。启动成功后可通过/v1/models接口可查看模型列表curlhttp://127.0.0.1:8060/v1/models-HAuthorization: Bearer token-abc123测试API交互思考模式curlhttp://127.0.0.1:8060/v1/chat/completions\-HContent-Type: application/json\-HAuthorization: Bearer token-abc123\-d{ model: Qwen3-32B, messages: [ {role: system, content: You are a helpful assistant.}, {role: user, content: 你是谁} ] }非思考模式测试curlhttp://127.0.0.1:8060/v1/chat/completions\-HContent-Type: application/json\-HAuthorization: Bearer token-abc123\-d{ model: Qwen3-32B, messages: [ {role: system, content: You are a helpful assistant.}, {role: user, content: 你是谁/no_think} ] }三、构建DB MCP Server在MCP Server端依据上面图片的规划包括三个MCP Tool分别是获取所有可用的表名get_all_tables、根据表名获取Schema get_table_schema、执行SQLrun_sql交互协议选择SSE模式。首先实现数据库操作这里仅仅做了数据库的交互实际使用你应考虑很多性能细节的优化utils_db.pyimportpymysqldefget_conn():returnpymysql.connect(host127.0.0.1,port3306,databaselangchain,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()returnresdb_mcp_server.pyimportjsonfrommcp.server.fastmcpimportFastMCPimportutils_db mcpFastMCP(DB Mcp Server)all_tables_sqlSELECT TABLE_NAME, TABLE_COMMENT FROM INFORMATION_SCHEMA.TABLES WHERE TABLE_SCHEMA langchainschema_sql SELECT COLUMN_NAME, DATA_TYPE, COLUMN_COMMENT FROM INFORMATION_SCHEMA.COLUMNS WHERE TABLE_SCHEMA langchain AND TABLE_NAME {table} mcp.tool()defget_all_tables()-str:获取所有可用的表名returnjson.dumps(utils_db.query(all_tables_sql),ensure_asciiFalse)mcp.tool()defget_table_schema(table_names:list[str])-str:根据表名获取Schematable_schema[]fortableintable_names:schemasutils_db.query(schema_sql.format(tabletable))schemas, \n.join([f{s[COLUMN_NAME]}{s[DATA_TYPE]}COMMENT{s[COLUMN_COMMENT]}forsinschemas])table_schema.append(f{table}({schemas}))return\n\n.join(table_schema)mcp.tool()defrun_sql(sql:str)-str:执行SQL查询数据一次仅能执行一句SQLtry:returnjson.dumps(utils_db.query(sql),ensure_asciiFalse)exceptExceptionase:returnf执行SQL错误{str(e)},请修正后重新发起。if__name____main__:mcp.settings.port6030mcp.run(sse)启动MCP Server服务四、Langchain 构建 MCP Client Agent 智能体官方关于MCP的集成介绍文档https://langchain-ai.github.io/langgraph/agents/mcp/实现过程importos,config os.environ[OPENAI_BASE_URL]http://127.0.0.1:8060/v1os.environ[OPENAI_API_KEY]token-abc123fromlangchain_mcp_adapters.clientimportMultiServerMCPClientfromlanggraph.prebuiltimportcreate_react_agentfromlanggraph.checkpoint.memoryimportInMemorySaverimportasynciofromcoloramaimportFore,Style,initasyncdefmain():clientMultiServerMCPClient({db:{url:http://127.0.0.1:6030/sse,transport:sse,}})toolsawaitclient.get_tools()checkpointerInMemorySaver()agentcreate_react_agent(openai:Qwen3-32B,tools,checkpointercheckpointer)config{configurable:{thread_id:1}}whileTrue:questioninput(请输入问题)ifnotquestion:continueifquestionq:breakasyncforchunkinagent.astream({messages:[{role:user,content:question}]},configconfig,stream_modeupdates):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())运行智能体开始测试。四、智能体问答测试提问当前的用户数以及工作组清单提问工作组 A 下的人都有哪些角色可以看到执行过程当发现错误后能够及时的纠正进而得到正确的结果提问role1 下的有哪些人