LangChain + MCP + vLLM + Qwen3-32B 构建本地私有化智能体应用
一、私有化智能体应用
在本专栏的前面文章基于Spring AI MCP实现了本地 ChatBI 问答应用，本文还是依据该场景，采用 LangChain + vLLM + Qwen3-32B + MCP 技术栈构建该流程，整体过程如下图所示：
实现效果如下所示：
实验所使用依赖的版本如下：
torch==2.6.0
transformers==4.51.3
modelscope==1.23.1
vllm==0.8.4
mcp==1.9.2
openai==1.75.0
langchain==0.3.25
langchain-openai==0.3.18
langgraph==0.4.7
pymysql==1.0.3
二、vLLM 部署 Qwen3-32B
使用 modelscope 下载 Qwen3-32B 模型到本地：
modelscope download --model="Qwen/Qwen3-32B" --local_dir Qwen3-32B
vLLM 读取模型启动API服务。
export CUDA_VISIBLE_DEVICES=0,1
vllm serve "Qwen3-32B" \
--host 0.0.0.0 \
--port 8060 \
--dtype bfloat16 \
--tensor-parallel-size 2 \
--cpu-offload-gb 0 \
--gpu-memory-utilization 0.8 \
--max-model-len 8126 \
--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_DEVICES=0,1：指定所使用的GPU。dtype: 数据类型，其中bfloat16，16位浮点数，适合NVIDIA A100等设备。tensor-parallel-size：Tensor并行的数量，当多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 接口可查看模型列表：
curl http://127.0.0.1:8060/v1/models -H "Authorization: Bearer token-abc123"
测试API交互，思考模式：
curl http://127.0.0.1:8060/v1/chat/completions \
-H "Content-Type: application/json" \
-H "Authorization: Bearer token-abc123" \
-d '{
"model": "Qwen3-32B",
"messages": [
{"role": "system", "content": "You are a helpful assistant."},
{"role": "user", "content": "你是谁"}
]
}'
非思考模式测试：
curl http://127.0.0.1:8060/v1/chat/completions \
-H "Content-Type: application/json" \
-H "Authorization: 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、执行SQL：run_sql ，交互协议选择 SSE 模式。
首先实现数据库操作，这里仅仅做了数据库的交互，实际使用你应考虑很多性能细节的优化：
utils_db.py
import pymysql
defget_conn():
return pymysql.connect(
host="127.0.0.1",
port=3306,
database="langchain",
user="root",
password="root",
autocommit=True
)
defquery(sql):
conn = get_conn()
cursor = conn.cursor()
cursor.execute(sql)
columns = [column[0] for column in cursor.description]
res = list()
for row in cursor.fetchall():
res.append(dict(zip(columns, row)))
cursor.close()
conn.close()
return res
db_mcp_server.py
import json
from mcp.server.fastmcp import FastMCP
import utils_db
mcp = FastMCP("DB Mcp Server")
all_tables_sql = "SELECT TABLE_NAME, TABLE_COMMENT FROM INFORMATION_SCHEMA.TABLES WHERE TABLE_SCHEMA = 'langchain'"
schema_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:
"""获取所有可用的表名"""
return json.dumps(utils_db.query(all_tables_sql), ensure_ascii=False)
@mcp.tool()
defget_table_schema(table_names: list[str]) -> str:
"""根据表名获取Schema"""
table_schema = []
for table in table_names:
schemas = utils_db.query(schema_sql.format(table=table))
schemas = ", \n".join([f"{s['COLUMN_NAME']} {s['DATA_TYPE']} COMMENT {s['COLUMN_COMMENT']}"for s in schemas])
table_schema.append(f"{table} ({schemas})")
return"\n\n".join(table_schema)
@mcp.tool()
defrun_sql(sql: str) -> str:
"""执行SQL查询数据，一次仅能执行一句SQL！"""
try:
return json.dumps(utils_db.query(sql), ensure_ascii=False)
except Exception as e:
returnf"执行SQL错误：{str(e)} ,请修正后重新发起。"
if __name__ == "__main__":
mcp.settings.port = 6030
mcp.run("sse")
启动 MCP Server 服务：
四、Langchain 构建 MCP Client Agent 智能体
官方关于 MCP 的集成介绍文档：
https://langchain-ai.github.io/langgraph/agents/mcp/
实现过程：
import os, config
os.environ["OPENAI_BASE_URL"] = "http://127.0.0.1:8060/v1"
os.environ["OPENAI_API_KEY"] = "token-abc123"
from langchain_mcp_adapters.client import MultiServerMCPClient
from langgraph.prebuilt import create_react_agent
from langgraph.checkpoint.memory import InMemorySaver
import asyncio
from colorama import Fore, Style, init
asyncdefmain():
client = MultiServerMCPClient(
{
"db": {
"url": "http://127.0.0.1:6030/sse",
"transport": "sse",
}
}
)
tools = await client.get_tools()
checkpointer = InMemorySaver()
agent = create_react_agent(
"openai:Qwen3-32B",
tools,
checkpointer=checkpointer
)
config = {
"configurable": {
"thread_id": "1"
}
}
whileTrue:
question = input("请输入问题：")
ifnot question:
continue
if question == "q":
break
asyncfor chunk in agent.astream(
{
"messages": [
{
"role": "user",
"content": question
}
]
},
config=config,
stream_mode="updates"
):
if"agent"in chunk:
content = chunk["agent"]["messages"][0].content
tool_calls = chunk["agent"]["messages"][0].tool_calls
if tool_calls:
for tool in tool_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"in chunk:
content = chunk["tools"]["messages"][0].content
name = chunk["tools"]["messages"][0].name
print(Fore.GREEN, Style.BRIGHT, f"<<< {name} : {content}")
if __name__ == '__main__':
asyncio.run(main())
运行智能体，开始测试。
四、智能体问答测试
提问：当前的用户数，以及工作组清单
提问：工作组 A 下的人，都有哪些角色
可以看到执行过程，当发现错误后，能够及时的纠正，进而得到正确的结果：
提问：role1 下的有哪些人
如何学习大模型 AI ？
由于新岗位的生产效率，要优于被取代岗位的生产效率，所以实际上整个社会的生产效率是提升的。
但是具体到个人，只能说是：
“最先掌握AI的人，将会比较晚掌握AI的人有竞争优势”。
这句话，放在计算机、互联网、移动互联网的开局时期，都是一样的道理。
我在一线互联网企业工作十余年里，指导过不少同行后辈。帮助很多人得到了学习和成长。
我意识到有很多经验和知识值得分享给大家，也可以通过我们的能力和经验解答大家在人工智能学习中的很多困惑，所以在工作繁忙的情况下还是坚持各种整理和分享。但苦于知识传播途径有限，很多互联网行业朋友无法获得正确的资料得到学习提升，故此将并将重要的AI大模型资料包括AI大模型入门学习思维导图、精品AI大模型学习书籍手册、视频教程、实战学习等录播视频免费分享出来。
第一阶段（10天）：初阶应用
该阶段让大家对大模型 AI有一个最前沿的认识，对大模型 AI 的理解超过 95% 的人，可以在相关讨论时发表高级、不跟风、又接地气的见解，别人只会和 AI 聊天，而你能调教 AI，并能用代码将大模型和业务衔接。
- 大模型 AI 能干什么？
- 大模型是怎样获得「智能」的？
- 用好 AI 的核心心法
- 大模型应用业务架构
- 大模型应用技术架构
- 代码示例：向 GPT-3.5 灌入新知识
- 提示工程的意义和核心思想
- Prompt 典型构成
- 指令调优方法论
- 思维链和思维树
- Prompt 攻击和防范
- …
第二阶段（30天）：高阶应用
该阶段我们正式进入大模型 AI 进阶实战学习，学会构造私有知识库，扩展 AI 的能力。快速开发一个完整的基于 agent 对话机器人。掌握功能最强的大模型开发框架，抓住最新的技术进展，适合 Python 和 JavaScript 程序员。
- 为什么要做 RAG
- 搭建一个简单的 ChatPDF
- 检索的基础概念
- 什么是向量表示（Embeddings）
- 向量数据库与向量检索
- 基于向量检索的 RAG
- 搭建 RAG 系统的扩展知识
- 混合检索与 RAG-Fusion 简介
- 向量模型本地部署
- …
第三阶段（30天）：模型训练
恭喜你，如果学到这里，你基本可以找到一份大模型 AI相关的工作，自己也能训练 GPT 了！通过微调，训练自己的垂直大模型，能独立训练开源多模态大模型，掌握更多技术方案。
到此为止，大概2个月的时间。你已经成为了一名“AI小子”。那么你还想往下探索吗？
- 为什么要做 RAG
- 什么是模型
- 什么是模型训练
- 求解器 & 损失函数简介
- 小实验2：手写一个简单的神经网络并训练它
- 什么是训练/预训练/微调/轻量化微调
- Transformer结构简介
- 轻量化微调
- 实验数据集的构建
- …
第四阶段（20天）：商业闭环
对全球大模型从性能、吞吐量、成本等方面有一定的认知，可以在云端和本地等多种环境下部署大模型，找到适合自己的项目/创业方向，做一名被 AI 武装的产品经理。
- 硬件选型
- 带你了解全球大模型
- 使用国产大模型服务
- 搭建 OpenAI 代理
- 热身：基于阿里云 PAI 部署 Stable Diffusion
- 在本地计算机运行大模型
- 大模型的私有化部署
- 基于 vLLM 部署大模型
- 案例：如何优雅地在阿里云私有部署开源大模型
- 部署一套开源 LLM 项目
- 内容安全
- 互联网信息服务算法备案
- …
学习是一个过程，只要学习就会有挑战。天道酬勤，你越努力，就会成为越优秀的自己。
如果你能在15天内完成所有的任务，那你堪称天才。然而，如果你能完成 60-70% 的内容，你就已经开始具备成为一名大模型 AI 的正确特征了。
这份完整版的大模型 AI 学习资料已经上传CSDN，朋友们如果需要可以微信扫描下方CSDN官方认证二维码免费领取【保证100%免费】
北京人形旗下天工造物具身智能开源社区，聚焦具身天工与慧思开物两大平台
更多推荐
15
0- 0
已为社区贡献44条内容
所有评论(0)