一、私有化智能体应用

在本专栏的前面文章基于Spring AI MCP实现了本地 ChatBI 问答应用，本文还是依据该场景，采用 LangChain + vLLM + Qwen3-32B + MCP 技术栈构建该流程，整体过程如下图所示：

实现效果如下所示：

关于 MySQL 表结构的创建，可以参考下面这篇文章：

Spring AI MCP Server + Cline 快速搭建一个数据库 ChatBi 助手

实验所使用依赖的版本如下：

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大模型的技术岗位与能力培养随着人工智能技术的迅速发展和应用 ， 大模型作为其中的重要组成部分 ， 正逐渐成为推动人工智能发展的重要引擎 。大模型以其强大的数据处理和模式识别能力， 广泛应用于自然语言处理 、计算机视觉 、 智能推荐等领域 ，为各行各业带来了革命性的改变和机遇 。

目前，开源人工智能大模型已应用于医疗、政务、法律、汽车、娱乐、金融、互联网、教育、制造业、企业服务等多个场景，其中，应用于金融、企业服务、制造业和法律领域的大模型在本次调研中占比超过 30%。

随着AI大模型技术的迅速发展，相关岗位的需求也日益增加。大模型产业链催生了一批高薪新职业：

人工智能大潮已来，不加入就可能被淘汰。如果你是技术人，尤其是互联网从业者，现在就开始学习AI大模型技术，真的是给你的人生一个重要建议！

最后

只要你真心想学习AI大模型技术，这份精心整理的学习资料我愿意无偿分享给你，但是想学技术去乱搞的人别来找我！

在当前这个人工智能高速发展的时代，AI大模型正在深刻改变各行各业。我国对高水平AI人才的需求也日益增长，真正懂技术、能落地的人才依旧紧缺。我也希望通过这份资料，能够帮助更多有志于AI领域的朋友入门并深入学习。

真诚无偿分享！！！
vx扫描下方二维码即可
加上后会一个个给大家发

大模型全套学习资料展示

自我们与MoPaaS魔泊云合作以来，我们不断打磨课程体系与技术内容，在细节上精益求精，同时在技术层面也新增了许多前沿且实用的内容，力求为大家带来更系统、更实战、更落地的大模型学习体验。

希望这份系统、实用的大模型学习路径，能够帮助你从零入门，进阶到实战，真正掌握AI时代的核心技能！

01 教学内容

• 从零到精通完整闭环：【基础理论 →RAG开发 → Agent设计 → 模型微调与私有化部署调→热门技术】5大模块，内容比传统教材更贴近企业实战！

• 大量真实项目案例： 带你亲自上手搞数据清洗、模型调优这些硬核操作，把课本知识变成真本事‌！

02适学人群

应届毕业生‌： 无工作经验但想要系统学习AI大模型技术，期待通过实战项目掌握核心技术。

零基础转型‌： 非技术背景但关注AI应用场景，计划通过低代码工具实现“AI+行业”跨界‌。

业务赋能突破瓶颈： 传统开发者（Java/前端等）学习Transformer架构与LangChain框架，向AI全栈工程师转型‌。

vx扫描下方二维码即可

本教程比较珍贵，仅限大家自行学习，不要传播！更严禁商用！

03 入门到进阶学习路线图

大模型学习路线图，整体分为5个大的阶段：

04 视频和书籍PDF合集

从0到掌握主流大模型技术视频教程（涵盖模型训练、微调、RAG、LangChain、Agent开发等实战方向）

新手必备的大模型学习PDF书单来了！全是硬核知识，帮你少走弯路（不吹牛，真有用）

05 行业报告+白皮书合集

收集70+报告与白皮书，了解行业最新动态！

06 90+份面试题/经验

AI大模型岗位面试经验总结（谁学技术不是为了赚$呢，找个好的岗位很重要）

07 deepseek部署包+技巧大全

由于篇幅有限

只展示部分资料

并且还在持续更新中…

真诚无偿分享！！！
vx扫描下方二维码即可
加上后会一个个给大家发