☰
隔离内网AI Agent工程实战:模型部署、MCP工具链与并发优化
2026/10/3 11:14:15 网站建设 项目流程

1. 隔离内网下的 AI Agent 工程,到底在解决什么问题

第一次听到“隔离内网”和“AI Agent”这两个词放在一起,很多人的第一反应是:这不是自相矛盾吗?Agent 要调模型、要访问外部工具、要拉取依赖,内网连外网都出不去,怎么玩?我刚开始接触这类需求时也是同样的疑问,直到真正在几个完全物理隔离的环境里把 Agent 从零跑起来,才发现这件事不但能做,而且做出来之后稳定性反而比公网环境更好——因为所有变量都被你捏在手里了。

先把概念理清楚。这里说的隔离内网,指的是没有直接互联网出口、或者只有极少数白名单出口的局域网环境,常见于金融、制造、能源、科研院所等对数据外流极度敏感的行业。这类环境里,开发机可能连 pip install 都跑不通,更别说调用云端大模型 API。而AI Agent 工程,指的是把大模型能力、工具调用、任务编排、记忆管理这一整套东西,做成一个能真正干活的生产系统,而不是停留在 demo 阶段的聊天窗口。

那这两者结合之后,核心矛盾就三个:模型从哪来、工具怎么接、依赖怎么装。公网环境下你随手一个 API Key 就能解决的事,在内网里要拆成十几个步骤,每一步都得有离线方案。我踩过的坑包括但不限于:模型权重下载了三天结果发现格式不对、MCP Server 的依赖树里有个包死活找不到离线 wheel、Agent 的向量库在内网机器上因为缺少某个系统库直接段错误。

这篇文章适合谁看?如果你正在或即将在隔离内网环境里落地 AI Agent,不管你是后端工程师、算法工程师还是技术负责人,这里面的思路和实操细节都能直接拿去用。如果你只是在公网环境玩 Agent,也可以看看内网约束下被迫做出的那些工程取舍,很多时候反而能帮你把系统设计得更健壮。接下来我会从整体设计思路开始,一步步拆到模型部署、MCP 工具链、Skills 编排、并发处理和问题排查,全部基于真实项目经验,不玩虚的。

2. 整体架构设计与技术选型思路

2.1 为什么内网 Agent 的架构必须“反着设计”

公网环境下做 Agent,大家的习惯是自顶向下:先选一个强大的云端模型,再挑几个好用的 SaaS 工具,最后用 LangChain 或者类似框架串起来。但内网环境必须反过来,自底向上:先看内网有什么硬件、什么系统、什么网络策略,再决定模型能跑多大、工具能接哪些、框架能用什么。

我总结了一个内网 Agent 架构的“三环模型”,从内到外分别是:

  • 算力环:GPU 型号、显存大小、CUDA 版本、驱动版本。这决定了你能跑什么量级的模型。比如两张 4090 加起来 48G 显存,7B 模型全量加载没问题,14B 量化后也能跑,但 70B 就别想了。
  • 依赖环:内网能访问的镜像源、能拷贝进来的离线包、系统里预装的 Python 版本和系统库。这一环最容易被低估,我见过太多项目卡在“就差一个 so 文件”上。
  • 协议环:Agent 和工具之间怎么通信。公网常用 HTTP + JSON Schema,内网里如果要做进程隔离,gRPC 或者本地 socket 反而更稳。MCP 协议在这里的价值就体现出来了,它本质上是一套标准化的工具描述和调用约定,只要双方都实现这个协议,不管底层是 HTTP 还是 stdio 都能对接。

提示:内网架构设计的第一原则是“假设所有外部依赖都不可用”,任何需要联网的环节都必须有离线替代方案,否则这个环节就是定时炸弹。

2.2 模型选型:不是越大越好,而是越“可控”越好

内网部署模型,第一个要放弃的执念就是“追新追大”。公网上今天出个新模型明天就想试,内网里你每换一次模型,意味着重新下载权重、重新验证显存占用、重新调 prompt 模板、重新跑一遍回归测试。这个成本极高。

我的选型逻辑是这样的:

场景推荐模型量级显存需求量化方式理由
简单工具调用、意图识别7B8-16GGPTQ-Int4响应快,显存友好,够用
复杂推理、多步规划14B-32B24-48GAWQ-Int4推理能力明显提升,仍可单卡
代码生成、长文档理解32B+48G+混合精度需要更大上下文窗口

这里有个关键决策点:用 Ollama 还是 vLLM。Ollama 部署简单,一条命令就能跑,但并发能力弱,适合个人开发或者低并发场景。vLLM 的 PagedAttention 机制对显存利用率和吞吐量提升非常明显,但部署复杂度高,需要匹配 CUDA 版本和 PyTorch 版本。我实测下来,同样一张 A100,vLLM 的并发吞吐大概是 Ollama 的 5-8 倍。如果你的 Agent 要服务多个用户或者多个并发任务,vLLM 是唯一选择。

2.3 MCP 协议在内网环境下的特殊价值

MCP(Model Context Protocol)这个词最近热度很高,但很多人对它的理解停留在“又一个工具调用协议”。在内网环境下,MCP 的真正价值在于解耦。

公网环境下,Agent 调工具直接写个函数就行,反正都在一个进程里。但内网环境往往有安全审计要求,工具的执行可能需要独立进程、独立权限、独立日志。MCP 定义了一套标准的 Server-Client 架构,Agent 作为 Client,工具作为 Server,两者通过 stdio 或者 SSE 通信。这意味着:

  • 工具 Server 可以用任何语言写,只要实现 MCP 协议
  • 工具 Server 可以跑在另一台机器上,通过网络通信
  • 工具的执行日志可以独立收集,方便审计
  • 工具的权限可以单独控制,比如数据库查询工具只能读不能写

我见过一个很典型的内网场景:Agent 需要查询内部知识库,但知识库有严格的访问控制,不能让 Agent 直接连数据库。这时候用 MCP 包一层,Server 端做权限校验和查询改写,Agent 只负责发请求,安全边界非常清晰。

2.4 Skills 机制:让 Agent 的能力可插拔

Skills 这个概念在不同框架里叫法不一样,Claude 叫 Agent Skills,Codex 也有类似机制,本质上都是把一组相关的工具调用、prompt 模板、执行逻辑打包成一个可复用的能力单元。

内网环境下 Skills 的价值更大,因为:

  • 能力可以离线分发,一个 skill 包拷贝进去就能用
  • 版本管理清晰,出问题可以快速回滚
  • 不同团队可以各自开发 skill,最后组装成完整 Agent

我目前的做法是每个 skill 一个独立目录,包含manifest.json(描述 skill 的元信息)、prompt.md(该 skill 专用的 prompt 模板)、tools/(该 skill 用到的工具定义)、tests/(回归测试用例)。这样整个 skill 可以打包成一个 zip,内网里解压即用。

3. 离线环境下的模型部署与依赖管理实操

3.1 模型权重下载与传输的完整流程

这是内网 Agent 工程的第一道坎,也是最容易翻车的地方。完整流程分四步:

第一步,在公网机器上确定模型版本和格式。不要直接下载,先确认内网推理框架支持什么格式。vLLM 支持 HuggingFace 格式和 GPTQ/AWQ 量化格式,Ollama 支持 GGUF 格式。格式不对,下载再多也白搭。

第二步,下载权重。用huggingface-cli download或者modelscope下载。这里有个技巧:如果模型很大,用--resume-download支持断点续传,并且优先选择 safetensors 格式,比 bin 格式加载快且安全。

# 示例:下载 Qwen2.5-14B-Instruct 的 GPTQ 量化版本 huggingface-cli download Qwen/Qwen2.5-14B-Instruct-GPTQ-Int4 \ --local-dir ./qwen2.5-14b-gptq \ --local-dir-use-symlinks False

第三步,传输到内网。根据内网的安全策略,可能是刻盘、可能是通过单向光闸、可能是通过指定的文件摆渡系统。不管哪种方式,传输前一定要做完整性校验。

# 生成校验文件 find ./qwen2.5-14b-gptq -type f -exec sha256sum {} \; > checksum.txt # 内网侧校验 sha256sum -c checksum.txt

第四步,内网侧加载验证。先跑一个最小推理脚本,确认模型能加载、能输出、显存占用符合预期。

from vllm import LLM, SamplingParams llm = LLM(model="/path/to/qwen2.5-14b-gptq", quantization="gptq", max_model_len=8192, gpu_memory_utilization=0.9) sampling_params = SamplingParams(temperature=0.1, max_tokens=512) outputs = llm.generate(["你好,请介绍一下你自己"], sampling_params) print(outputs[0].outputs[0].text)

注意:内网机器上的 CUDA 版本必须和 vLLM 编译时的 CUDA 版本匹配,否则会出现莫名其妙的 kernel 错误。我遇到过 CUDA 12.1 的 vLLM 跑在 CUDA 11.8 驱动上,加载模型直接 core dump,排查了一整天。

3.2 Python 依赖的离线安装方案

内网里pip install基本废了,必须提前准备好离线包。我的标准做法是:

方案一:pip download 全量打包。在公网机器上,用和目标机器相同的 Python 版本和操作系统,下载所有依赖的 wheel 文件。

# 下载依赖及其所有子依赖 pip download -r requirements.txt -d ./offline_packages \ --python-version 310 \ --platform manylinux2014_x86_64 \ --only-binary=:all:

这里的关键是--platform和--python-version必须和目标环境完全一致,否则下载的 wheel 装不上。如果有些包没有预编译 wheel,需要加--no-binary并确保目标机器有编译环境。

方案二:conda-pack 整包迁移。如果内网机器连编译环境都没有,用 conda-pack 把整个虚拟环境打包。

conda pack -n my_agent_env -o agent_env.tar.gz # 内网侧解压 mkdir -p /opt/agent_env tar -xzf agent_env.tar.gz -C /opt/agent_env source /opt/agent_env/bin/activate

这个方案的好处是连 Python 解释器都打包进去了,完全不依赖系统环境。缺点是包体积大,通常 2-5G。

方案三:Docker 镜像离线导入。如果内网支持容器,这是最干净的方案。

# 公网侧构建并导出 docker build -t agent-runtime:v1 . docker save agent-runtime:v1 -o agent-runtime-v1.tar # 内网侧导入 docker load -i agent-runtime-v1.tar

我个人的优先级是:Docker > conda-pack > pip download。Docker 的隔离性最好,但需要内网有容器运行时;conda-pack 最灵活,适合没有容器环境的场景。

3.3 向量库和嵌入模型的离线部署

Agent 要做 RAG 或者长期记忆,向量库和嵌入模型是绕不开的。内网环境下推荐两个方案:

轻量方案:ChromaDB + BGE-small-zh。ChromaDB 可以纯本地运行,不需要额外服务。BGE-small-zh 模型只有 100M 左右,CPU 推理也很快。

from chromadb import Client from sentence_transformers import SentenceTransformer embed_model = SentenceTransformer("/path/to/bge-small-zh") client = Client(path="/data/chroma_db") collection = client.create_collection("knowledge_base") # 嵌入并入库 texts = ["文档1内容", "文档2内容"] embeddings = embed_model.encode(texts).tolist() collection.add(documents=texts, embeddings=embeddings, ids=["1", "2"])

生产方案:Milvus 单机版 + BGE-large-zh。Milvus 支持更大规模的数据和更高的并发查询,但部署复杂度高,需要 etcd、MinIO 等组件。如果数据量在百万级以下,ChromaDB 完全够用。

嵌入模型的选择上,中文场景我强烈推荐 BGE 系列,它在中文语义相似度任务上的表现明显优于同量级的其他开源模型。如果内网机器有 GPU,可以用 BGE-large-zh 获得更好的效果;纯 CPU 环境用 BGE-small-zh 就够了。

4. MCP 工具链与 Skills 编排的内网落地

4.1 MCP Server 的离线开发与部署

MCP Server 本质上就是一个实现了特定协议的进程,它可以是 Python 脚本、Node.js 服务,甚至是一个编译好的二进制文件。内网部署 MCP Server 的关键在于依赖自包含。

我通常用 Python 写 MCP Server,然后用 PyInstaller 打包成单文件可执行程序,这样内网机器上不需要 Python 环境就能跑。

# mcp_server_example.py from mcp.server import Server, NotificationOptions from mcp.server.models import InitializationOptions import mcp.server.stdio import mcp.types as types server = Server("internal-tools") @server.list_tools() async def handle_list_tools() -> list[types.Tool]: return [ types.Tool( name="query_knowledge_base", description="查询内部知识库", inputSchema={ "type": "object", "properties": { "query": {"type": "string", "description": "查询语句"} }, "required": ["query"] } ) ] @server.call_tool() async def handle_call_tool(name: str, arguments: dict) -> list[types.TextContent]: if name == "query_knowledge_base": result = search_kb(arguments["query"]) return [types.TextContent(type="text", text=result)] raise ValueError(f"Unknown tool: {name}") async def main(): async with mcp.server.stdio.stdio_server() as (read_stream, write_stream): await server.run(read_stream, write_stream, InitializationOptions( server_name="internal-tools", server_version="0.1.0" )) if __name__ == "__main__": import asyncio asyncio.run(main())

打包命令:

pyinstaller --onefile --name mcp-server mcp_server_example.py

打包出来的单文件大概 15-30M,拷贝到内网直接运行。Agent 侧配置 MCP Client 连接这个 Server 即可。

提示:PyInstaller 打包时要注意隐藏导入的问题,有些库是动态加载的,PyInstaller 分析不到。用--hidden-import手动指定,或者用--collect-all把整个包打进去。

4.2 Skills 的设计模式与内网分发

Skills 的设计我遵循“单一职责”原则,一个 skill 只做一件事,但要把这件事做透。比如“查询数据库”是一个 skill,“生成报表”是另一个 skill,“发送通知”又是一个 skill。这样组合起来灵活,出问题也好定位。

一个完整的 skill 目录结构:

skills/ query_database/ manifest.json prompt.md tools/ query_tool.py tests/ test_query.py requirements.txt

manifest.json描述 skill 的元信息:

{ "name": "query_database", "version": "1.0.0", "description": "查询内部数据库并返回结构化结果", "author": "team-data", "dependencies": ["pymysql>=1.0.0"], "tools": ["query_tool"], "prompt_template": "prompt.md" }

内网分发时,把整个 skills 目录打包,通过文件摆渡系统拷贝进去,Agent 启动时扫描 skills 目录自动加载。我一般会写一个skill_loader.py:

import json import os from pathlib import Path def load_skills(skills_dir: str): skills = [] for skill_path in Path(skills_dir).iterdir(): if not skill_path.is_dir(): continue manifest_file = skill_path / "manifest.json" if not manifest_file.exists(): continue with open(manifest_file, "r", encoding="utf-8") as f: manifest = json.load(f) manifest["path"] = str(skill_path) skills.append(manifest) return skills

4.3 Agent 编排:从单步调用到多步规划

内网 Agent 的编排逻辑和公网没有本质区别,但因为模型能力可能弱一些(毕竟跑不了 GPT-4 级别的模型),所以 prompt 工程和流程设计要更精细。

我的做法是把 Agent 的执行流程拆成三个阶段:

阶段一:意图理解。用一个小模型或者规则引擎做意图分类,判断用户请求属于哪个 skill 的范畴。这一步不需要很强的模型,7B 甚至更小就够。

阶段二:任务规划。用主模型做多步规划,把复杂任务拆成子任务序列。这里的关键是给模型足够的上下文,包括可用 skill 列表、每个 skill 的输入输出格式、历史执行结果。

阶段三:执行与反思。按规划逐步执行,每步执行后检查结果是否符合预期,不符合则触发重试或者重新规划。

class AgentOrchestrator: def __init__(self, llm, skills, max_steps=10): self.llm = llm self.skills = {s["name"]: s for s in skills} self.max_steps = max_steps def run(self, user_input: str): # 阶段一:意图理解 intent = self.classify_intent(user_input) # 阶段二:任务规划 plan = self.plan_tasks(user_input, intent) # 阶段三:执行与反思 results = [] for step in plan[:self.max_steps]: result = self.execute_step(step) results.append(result) if not self.validate_result(step, result): plan = self.replan(user_input, results) return self.summarize(results)

注意:内网模型的规划能力通常弱于云端大模型,所以max_steps不要设太大,否则容易陷入死循环。我一般设 5-8 步,超过就强制返回当前结果并提示用户任务过于复杂。

5. 并发处理与性能调优的实战经验

5.1 AI Agent 怎么扛并发:从单线程到异步架构

“AI Agent 怎么扛并发”是最近被问得最多的问题之一。公网环境下你可以靠 API 的弹性扩容,内网里 GPU 就那么多,必须精打细算。

先看瓶颈在哪。一个 Agent 请求的耗时分布大概是:模型推理占 70%-90%,工具调用占 5%-15%,编排逻辑占 5%-10%。所以并发优化的核心是模型推理的并发。

vLLM 本身支持 continuous batching,多个请求会自动合并成一个 batch 推理,吞吐量提升非常明显。但前提是你的 Agent 框架要支持异步调用。

import asyncio from vllm import AsyncLLMEngine, SamplingParams from vllm.engine.arg_utils import AsyncEngineArgs engine_args = AsyncEngineArgs( model="/path/to/model", max_num_seqs=16, # 最大并发序列数 gpu_memory_utilization=0.9 ) engine = AsyncLLMEngine.from_engine_args(engine_args) async def generate(prompt: str): sampling_params = SamplingParams(temperature=0.1, max_tokens=512) results = [] async for output in engine.generate(prompt, sampling_params, request_id=prompt[:10]): results.append(output) return results[-1]

max_num_seqs是关键参数,它决定了同时处理多少个请求。设太小浪费显存,设太大容易 OOM。我的经验值是:7B 模型设 16-32,14B 模型设 8-16,32B 模型设 4-8。具体还要看显存大小和序列长度。

5.2 请求队列与优先级管理

内网 Agent 往往要服务多个业务方,不同业务方的请求优先级不同。这时候需要一个请求队列来管理。

我用 Redis 做队列(内网部署一个单机 Redis 即可),按优先级分三个队列:

  • 高优先级:实时交互请求,比如客服机器人,要求 3 秒内响应
  • 中优先级:批处理任务,比如文档分析,可以等 30 秒
  • 低优先级:后台任务,比如数据同步,可以等几分钟
import redis import json r = redis.Redis(host="localhost", port=6379) def enqueue_request(request, priority="medium"): queue_name = f"agent_queue:{priority}" r.lpush(queue_name, json.dumps(request)) def dequeue_request(): # 按优先级顺序取 for priority in ["high", "medium", "low"]: result = r.rpop(f"agent_queue:{priority}") if result: return json.loads(result) return None

Worker 进程从队列取任务,调用 Agent 执行,结果写回另一个队列或者直接回调。

5.3 缓存策略:减少重复推理

Agent 场景下有很多重复请求,比如相同的意图识别、相同的知识库查询。加一层缓存能显著降低 GPU 负载。

我的缓存分两层:

第一层:精确匹配缓存。用户输入完全相同时,直接返回缓存结果。用 Redis 的 String 类型,key 是输入文本的 hash,value 是结果 JSON。

import hashlib def get_cache_key(text: str) -> str: return f"agent:cache:{hashlib.md5(text.encode()).hexdigest()}" def get_cached_result(text: str): return r.get(get_cache_key(text)) def set_cached_result(text: str, result: str, ttl=3600): r.setex(get_cache_key(text), ttl, result)

第二层:语义缓存。用户输入不同但语义相同时,也能命中缓存。用嵌入模型把输入转成向量,在向量库里查相似度超过阈值的记录。

def get_semantic_cache(text: str, threshold=0.95): embedding = embed_model.encode(text).tolist() results = collection.query(query_embeddings=[embedding], n_results=1) if results["distances"][0][0] < (1 - threshold): return results["documents"][0][0] return None

语义缓存的命中率取决于阈值设置,太高了命中少,太低了容易返回不相关的结果。我实测下来 0.92-0.95 是比较平衡的区间。

6. 常见问题与排查技巧实录

6.1 模型加载失败类问题

现象可能原因排查方法解决方案
加载时 core dumpCUDA 版本不匹配nvcc --version和python -c "import torch; print(torch.version.cuda)"对比重新编译 vLLM 或降级 CUDA
显存不足 OOM模型太大或gpu_memory_utilization太高nvidia-smi观察显存占用降低max_model_len或换量化版本
加载极慢磁盘 IO 瓶颈iostat -x 1看磁盘利用率把模型放到 SSD 或内存盘
输出乱码tokenizer 不匹配检查 tokenizer 配置和模型是否配套使用模型自带的 tokenizer

6.2 MCP 连接类问题

MCP Server 连不上是最常见的问题,排查思路:

  1. 确认 Server 进程在跑:ps aux | grep mcp-server
  2. 确认通信方式:stdio 模式下 Server 必须由 Client 启动,不能手动启动;SSE 模式下检查端口是否监听
  3. 确认协议版本:Client 和 Server 的 MCP 协议版本要兼容,版本不匹配会握手失败
  4. 看日志:MCP Server 的 stderr 输出通常有详细错误信息

我遇到过一个很隐蔽的问题:MCP Server 用 PyInstaller 打包后,stdio 通信正常,但 SSE 模式死活连不上。最后发现是 PyInstaller 打包时把uvicorn的某些动态导入漏了,导致 SSE 服务启动失败。解决方案是加--collect-all uvicorn。

6.3 内网环境特有的坑

坑一:DNS 解析问题。内网机器可能没有配置 DNS,导致所有域名解析失败。如果 Agent 代码里有任何地方用了域名(比如连接内部服务),要么改成 IP,要么在/etc/hosts里加静态解析。

坑二:时间不同步。内网机器如果长时间不联网,系统时间可能漂移,导致 HTTPS 证书校验失败、JWT token 过期判断错误等。部署前务必用 NTP 同步时间,或者至少手动校准。

坑三:文件句柄限制。内网服务器默认的ulimit -n可能只有 1024,Agent 并发高了之后会出现 "Too many open files"。提前改大:

ulimit -n 65535 # 永久生效写入 /etc/security/limits.conf

坑四:日志磁盘写满。Agent 的日志量很大,尤其是 debug 级别。内网机器磁盘通常不大,很容易写满。我的做法是日志按天切割,保留 7 天,并且把日志级别默认设为 INFO。

import logging from logging.handlers import TimedRotatingFileHandler handler = TimedRotatingFileHandler( "/var/log/agent/agent.log", when="midnight", interval=1, backupCount=7, encoding="utf-8" ) handler.setFormatter(logging.Formatter( "%(asctime)s - %(name)s - %(levelname)s - %(message)s" )) logger = logging.getLogger("agent") logger.addHandler(handler) logger.setLevel(logging.INFO)

6.4 性能调优速查表

问题调优方向具体操作预期效果
响应慢模型推理换量化版本、减小 max_tokens提速 30%-50%
并发低批处理调大 max_num_seqs吞吐提升 3-5 倍
重复计算多缓存加精确缓存和语义缓存GPU 负载降低 40%
工具调用慢连接池数据库连接池、HTTP 连接复用工具耗时降低 50%
内存泄漏对象回收定期重启 Worker、检查循环引用稳定性提升

7. 一些个人体会和后续扩展方向

在内网里折腾 AI Agent 这一年多,最大的感受是:约束反而催生了更好的工程实践。公网环境下你可以随意试错,但内网里每一步都要想清楚,这种“被迫严谨”让整个系统的健壮性上了一个台阶。我现在回到公网环境做项目,也会沿用内网的很多做法,比如依赖锁定、离线包管理、严格的版本控制。

如果要把这套方案继续扩展,我觉得有几个方向值得投入:一是把 Skills 做成一个内部市场,不同团队开发的 skill 可以互相复用,减少重复造轮子;二是把 Agent 的执行轨迹做成可视化面板,方便排查问题和优化 prompt;三是探索小模型加规则引擎的混合方案,在保证效果的前提下进一步降低算力需求。

最后分享一个我踩过的最大的坑:不要在内网里用最新版本的任何东西。最新版的 vLLM、最新版的 PyTorch、最新版的 MCP SDK,在内网里出问题的概率远高于稳定版。我的原则是,公网社区验证过至少三个月、有大量生产案例的版本,才考虑引入内网。这个原则帮我省了至少几十个小时的排查时间。

需要专业的网站建设服务?

联系我们获取免费的网站建设咨询和方案报价,让我们帮助您实现业务目标

立即咨询