ARTICLE DETAIL

资讯详情

深耕郑州网站建设与运营推广的一线实战洞察。

隔离内网AI Agent工程实战:模型部署、MCP工具链与并发优化

隔离内网AI Agent工程实战:模型部署、MCP工具链与并发优化 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推理能力明显提升仍可单卡代码生成、长文档理解32B48G混合精度需要更大上下文窗口这里有个关键决策点用 Ollama 还是 vLLM。Ollama 部署简单一条命令就能跑但并发能力弱适合个人开发或者低并发场景。vLLM 的 PagedAttention 机制对显存利用率和吞吐量提升非常明显但部署复杂度高需要匹配 CUDA 版本和 PyTorch 版本。我实测下来同样一张 A100vLLM 的并发吞吐大概是 Ollama 的 5-8 倍。如果你的 Agent 要服务多个用户或者多个并发任务vLLM 是唯一选择。2.3 MCP 协议在内网环境下的特殊价值MCPModel 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 SkillsCodex 也有类似机制本质上都是把一组相关的工具调用、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, quantizationgptq, max_model_len8192, gpu_memory_utilization0.9) sampling_params SamplingParams(temperature0.1, max_tokens512) 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(documentstexts, embeddingsembeddings, 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( namequery_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(typetext, textresult)] raise ValueError(fUnknown 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_nameinternal-tools, server_version0.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.txtmanifest.json描述 skill 的元信息{ name: query_database, version: 1.0.0, description: 查询内部数据库并返回结构化结果, author: team-data, dependencies: [pymysql1.0.0], tools: [query_tool], prompt_template: prompt.md }内网分发时把整个 skills 目录打包通过文件摆渡系统拷贝进去Agent 启动时扫描 skills 目录自动加载。我一般会写一个skill_loader.pyimport 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, encodingutf-8) as f: manifest json.load(f) manifest[path] str(skill_path) skills.append(manifest) return skills4.3 Agent 编排从单步调用到多步规划内网 Agent 的编排逻辑和公网没有本质区别但因为模型能力可能弱一些毕竟跑不了 GPT-4 级别的模型所以 prompt 工程和流程设计要更精细。我的做法是把 Agent 的执行流程拆成三个阶段阶段一意图理解。用一个小模型或者规则引擎做意图分类判断用户请求属于哪个 skill 的范畴。这一步不需要很强的模型7B 甚至更小就够。阶段二任务规划。用主模型做多步规划把复杂任务拆成子任务序列。这里的关键是给模型足够的上下文包括可用 skill 列表、每个 skill 的输入输出格式、历史执行结果。阶段三执行与反思。按规划逐步执行每步执行后检查结果是否符合预期不符合则触发重试或者重新规划。class AgentOrchestrator: def __init__(self, llm, skills, max_steps10): 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_seqs16, # 最大并发序列数 gpu_memory_utilization0.9 ) engine AsyncLLMEngine.from_engine_args(engine_args) async def generate(prompt: str): sampling_params SamplingParams(temperature0.1, max_tokens512) results [] async for output in engine.generate(prompt, sampling_params, request_idprompt[:10]): results.append(output) return results[-1]max_num_seqs是关键参数它决定了同时处理多少个请求。设太小浪费显存设太大容易 OOM。我的经验值是7B 模型设 16-3214B 模型设 8-1632B 模型设 4-8。具体还要看显存大小和序列长度。5.2 请求队列与优先级管理内网 Agent 往往要服务多个业务方不同业务方的请求优先级不同。这时候需要一个请求队列来管理。我用 Redis 做队列内网部署一个单机 Redis 即可按优先级分三个队列高优先级实时交互请求比如客服机器人要求 3 秒内响应中优先级批处理任务比如文档分析可以等 30 秒低优先级后台任务比如数据同步可以等几分钟import redis import json r redis.Redis(hostlocalhost, port6379) def enqueue_request(request, prioritymedium): queue_name fagent_queue:{priority} r.lpush(queue_name, json.dumps(request)) def dequeue_request(): # 按优先级顺序取 for priority in [high, medium, low]: result r.rpop(fagent_queue:{priority}) if result: return json.loads(result) return NoneWorker 进程从队列取任务调用 Agent 执行结果写回另一个队列或者直接回调。5.3 缓存策略减少重复推理Agent 场景下有很多重复请求比如相同的意图识别、相同的知识库查询。加一层缓存能显著降低 GPU 负载。我的缓存分两层第一层精确匹配缓存。用户输入完全相同时直接返回缓存结果。用 Redis 的 String 类型key 是输入文本的 hashvalue 是结果 JSON。import hashlib def get_cache_key(text: str) - str: return fagent: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, ttl3600): r.setex(get_cache_key(text), ttl, result)第二层语义缓存。用户输入不同但语义相同时也能命中缓存。用嵌入模型把输入转成向量在向量库里查相似度超过阈值的记录。def get_semantic_cache(text: str, threshold0.95): embedding embed_model.encode(text).tolist() results collection.query(query_embeddings[embedding], n_results1) 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 配置和模型是否配套使用模型自带的 tokenizer6.2 MCP 连接类问题MCP Server 连不上是最常见的问题排查思路确认 Server 进程在跑ps aux | grep mcp-server确认通信方式stdio 模式下 Server 必须由 Client 启动不能手动启动SSE 模式下检查端口是否监听确认协议版本Client 和 Server 的 MCP 协议版本要兼容版本不匹配会握手失败看日志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可能只有 1024Agent 并发高了之后会出现 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, whenmidnight, interval1, backupCount7, encodingutf-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在内网里出问题的概率远高于稳定版。我的原则是公网社区验证过至少三个月、有大量生产案例的版本才考虑引入内网。这个原则帮我省了至少几十个小时的排查时间。
返回列表