1. 为什么“隔离内网 AI Agent”是个真问题先把场景说清楚。所谓隔离内网就是一台或者一批机器物理上或者策略上跟公网断开装不了外网包连不上外部模型 API甚至连 pip、npm 这类包管理器的默认源都访问不了。很多做 AI Agent 的朋友第一反应是这玩意儿离了公网还能跑吗答案是能跑而且跑得还挺好前提是你得把整套工程思路从“在线优先”切换成“离线自洽”。我最早接触这类需求是帮一个做工业质检的团队搭一套内部知识问答 Agent。他们的产线网络跟办公网是隔离的办公网又跟公网隔离三层结构。一开始我按常规思路用云端大模型 API 在线向量库结果连第一行代码都跑不起来——机器根本出不去。后来整套方案推倒重来改成“本地模型 本地向量库 本地工具链 内网自建服务”才真正落地。这里面的核心矛盾在于AI Agent 的工程实践天然依赖三类外部资源——模型推理能力、工具调用能力也就是常说的 MCP、Skills 这类机制、以及依赖包和运行时。隔离内网把这三条路全掐了。所以你要做的不是“想办法连出去”而是“把这三样东西全部搬进来并且让它们在内网里自己转起来”。这篇文章面向的是这样几类人一是在内网环境做 AI 应用落地的工程师二是想搞清楚 Agent 工程到底由哪些模块拼起来的技术负责人三是手里有内网服务器、想自己搭一套能用的 Agent 平台的折腾型选手。我会把模型部署、MCP 协议落地、Skills 机制设计、并发扛压、依赖离线化这几块拆开讲每一块都给可复现的思路和踩过的坑。提示本文所有方案都基于“完全离线、不依赖任何外部网络”的前提设计。如果你的内网其实有受控的出口那选择会多很多但本文不讨论那类场景。2. 把模型搬进内网选型、量化与推理服务2.1 内网模型选型的三条硬约束在公网环境选模型你主要看效果和价格。在内网选模型约束条件完全变了我总结成三条硬约束第一是显存约束。内网服务器往往是几年前采购的显卡可能是 V100、T4甚至只有 CPU。你得先摸清楚手上有多少显存再倒推能跑多大的模型。经验值是FP16 精度下模型参数量乘以 2 就是大致显存需求单位 GB7B 模型约 14GB13B 约 26GB再算上 KV Cache 和框架开销实际要留 20% 余量。第二是量化容忍度。内网机器不够强就得量化。INT8 量化大概能把显存砍一半INT4 再砍一半但效果会掉。我的经验是做知识问答和工具调用这类任务INT4 的 Qwen 系列、GLM 系列还能用但做复杂推理就会明显退化。所以量化策略要跟任务难度匹配。第三是许可证与合规。内网环境往往对许可证敏感选模型时要确认权重可以内部使用。这一点很多人会忽略等到部署完了才发现有问题返工成本很高。下面这张表是我实际用过的几个模型在内网场景下的对比供参考模型参数量INT4 显存占用工具调用能力中文表现内网适配建议Qwen2.5-7B-Instruct7B约 6GB较强优秀首选均衡Qwen2.5-14B-Instruct14B约 11GB强优秀显存够就上GLM-4-9B-Chat9B约 8GB较强优秀备选Llama3.1-8B-Instruct8B约 7GB中等一般英文场景Baichuan2-13B13B约 10GB中等良好老项目兼容2.2 推理框架怎么选vLLM、llama.cpp 还是 Ollama内网部署推理服务主流就三个选择各有适用场景。vLLM适合有像样显卡、要扛并发的场景。它的 PagedAttention 机制对显存利用率很高吞吐量比朴素实现高好几倍。缺点是安装依赖比较重离线安装要提前把 wheel 包和 CUDA 相关依赖全部下好。我一般用pip download在联网机器上把依赖拉全再拷进内网。llama.cpp适合 CPU 或者显存很小的场景。它支持 GGUF 格式的量化模型CPU 上也能跑出可用的速度。缺点是并发能力弱适合个人用或者低并发场景。它的离线部署最省心一个二进制文件加一个模型文件就能跑。Ollama是 llama.cpp 的封装用起来最舒服一条命令拉模型。但它的模型拉取默认走公网内网要手动把模型文件放进去或者自建一个模型仓库。它的 API 兼容 OpenAI 格式这点对 Agent 工程很友好。我的建议是有显卡、要扛并发上 vLLM纯 CPU 或者小显存用 llama.cpp想快速验证、不想折腾用 Ollama。三者都可以通过 OpenAI 兼容接口暴露出来这样上层 Agent 代码不用改。2.3 离线部署推理服务的完整链路以 vLLM 为例讲一下内网离线部署的完整链路。这套流程我在多个内网环境跑通过。第一步在联网机器上准备依赖。建一个干净的虚拟环境执行pip download vllm -d ./vllm_packages --platform manylinux2014_x86_64 --python-version 310 --only-binary:all:这一步会把 vllm 及其所有依赖的 wheel 包下到vllm_packages目录。注意--platform和--python-version要跟内网机器的环境对齐否则装不上。第二步把模型权重下好。从模型仓库把权重文件完整下载包括 config、tokenizer、safetensors 等所有文件。这一步最容易出错的是漏文件建议下载后用huggingface-cli的校验功能确认完整性。第三步把 wheel 包和模型文件拷进内网。用移动硬盘或者内网文件服务器都行。第四步在内网机器上安装pip install --no-index --find-links./vllm_packages vllm--no-index表示不走任何在线源--find-links指向本地包目录。这一步如果报缺依赖说明第一步没下全回去补。第五步启动推理服务python -m vllm.entrypoints.openai.api_server \ --model /path/to/model \ --served-model-name local-model \ --host 0.0.0.0 \ --port 8000 \ --max-model-len 8192 \ --gpu-memory-utilization 0.9启动后内网其他机器就能通过http://内网IP:8000/v1访问接口格式跟 OpenAI 一致。这样上层 Agent 代码里只要把 base_url 改掉就行。注意--gpu-memory-utilization别设成 1.0留一点给系统和其他进程否则容易 OOM。我一般设 0.85 到 0.9。2.4 模型服务的内网高可用考虑内网环境虽然流量不大但 Agent 一旦跑起来模型服务就是单点。我的做法是至少起两个实例前面挂一个内网 Nginx 做负载均衡。Nginx 配置很简单upstream model_backend { server 192.168.1.10:8000; server 192.168.1.11:8000; } server { listen 8080; location /v1/ { proxy_pass http://model_backend; proxy_read_timeout 300s; } }proxy_read_timeout要设大因为大模型推理有时候要几十秒默认 60 秒会断。这个坑我踩过Agent 调用到一半连接被切断排查了半天才发现是 Nginx 超时。3. MCP 协议在内网怎么落地3.1 MCP 到底解决了什么问题MCP 全称 Model Context Protocol你可以把它理解成“Agent 和外部工具之间的标准插座”。在没有 MCP 之前每接一个工具就要写一套适配代码工具多了就是一团乱麻。MCP 把这件事标准化了工具方实现一个 MCP ServerAgent 方实现一个 MCP Client双方通过统一的协议通信谁也不用关心对方内部怎么实现。在内网环境MCP 的价值反而更大。因为内网工具往往是自研的、五花八门的有查数据库的、有调内部 API 的、有操作文件的。如果每个都单独适配维护成本极高。用 MCP 统一起来新增工具只要起一个 ServerAgent 侧几乎不用改。MCP 的通信方式主要有两种stdio标准输入输出和 SSEServer-Sent Events。stdio 适合本地进程Agent 直接拉起一个子进程通信SSE 适合远程服务通过 HTTP 长连接通信。内网环境两种都能用我一般本地工具用 stdio跨机器的用 SSE。3.2 内网 MCP Server 的部署模式内网部署 MCP Server我总结出三种模式按复杂度递增。模式一单机 stdio 模式。所有 MCP Server 跟 Agent 跑在同一台机器上通过 stdio 通信。这种最简单适合工具少、单机部署的场景。缺点是工具跟 Agent 绑死没法复用。模式二内网 SSE 服务模式。把 MCP Server 做成独立的 HTTP 服务部署在内网某台机器上Agent 通过 SSE 连接。这种模式工具可以复用多个 Agent 共享同一批工具。缺点是要处理服务发现和鉴权。模式三MCP 网关模式。在内网起一个 MCP 网关所有 MCP Server 注册到网关Agent 只跟网关打交道。网关负责路由、鉴权、限流、日志。这种模式最适合工具体系庞大的场景但实现成本也最高。我的建议是刚开始用模式一快速验证工具超过五个就切模式二工具超过二十个或者有多个 Agent 共享需求再上模式三。3.3 一个内网 MCP Server 的最小实现下面给一个基于 Python 的 MCP Server 最小实现功能是查询内网数据库。这个例子我在实际项目里用过可以直接改。from mcp.server import Server from mcp.server.stdio import stdio_server from mcp.types import Tool, TextContent import sqlite3 app Server(internal-db-server) app.list_tools() async def list_tools(): return [ Tool( namequery_employee, description根据员工姓名查询员工信息, inputSchema{ type: object, properties: { name: {type: string, description: 员工姓名} }, required: [name] } ) ] app.call_tool() async def call_tool(name: str, arguments: dict): if name query_employee: conn sqlite3.connect(/data/employee.db) cursor conn.cursor() cursor.execute(SELECT * FROM employees WHERE name ?, (arguments[name],)) rows cursor.fetchall() conn.close() return [TextContent(typetext, textstr(rows))] raise ValueError(fUnknown tool: {name}) async def main(): async with stdio_server() as (read, write): await app.run(read, write, app.create_initialization_options()) if __name__ __main__: import asyncio asyncio.run(main())这个 Server 暴露了一个query_employee工具Agent 调用时传入姓名返回员工信息。实际项目里你可以把 sqlite 换成内网的 MySQL、PostgreSQL逻辑一样。3.4 MCP 在内网的鉴权与安全内网不等于安全这点必须强调。我见过太多内网系统裸奔任何能连上内网的人都能调。MCP Server 如果暴露了敏感操作比如删数据、发消息一定要加鉴权。最简单的做法是在 SSE 模式下加一个 Token 校验。Agent 连接时带上 TokenServer 校验通过才响应。Token 可以写死在配置里也可以走内网的统一认证服务。from fastapi import FastAPI, Request, HTTPException from mcp.server.sse import SseServerTransport app FastAPI() sse SseServerTransport(/messages) VALID_TOKEN internal-secret-token app.get(/sse) async def handle_sse(request: Request): token request.headers.get(Authorization) if token ! fBearer {VALID_TOKEN}: raise HTTPException(status_code401, detailUnauthorized) async with sse.connect_sse(request.scope, request.receive, request._send) as streams: await mcp_app.run(streams[0], streams[1], mcp_app.create_initialization_options())注意Token 别硬编码在代码里提交到版本库用环境变量或者内网配置中心注入。这个习惯在内网环境同样重要因为内网代码也可能被审计。4. Skills 机制让 Agent 真正“会干活”4.1 Skills 和 MCP 的区别别再搞混了很多人把 Skills 和 MCP 混为一谈其实两者定位完全不同。MCP 解决的是“Agent 怎么调用工具”的通信问题是连接层。Skills 解决的是“Agent 在什么场景下、按什么步骤、用什么工具”的编排问题是能力层。打个比方MCP 是插座和电线Skills 是“怎么用电饭煲煮饭”的菜谱。插座再多没有菜谱你也煮不出饭。反过来菜谱再详细没有插座你也用不了电饭煲。一个 Skill 通常包含几部分触发条件什么时候用这个 Skill、执行步骤分几步做、用到的工具通过 MCP 调用、输出格式结果长什么样。在内网环境Skills 往往是团队沉淀下来的“最佳实践”把老师傅的经验固化下来让 Agent 照着做。4.2 内网 Skills 的设计原则内网 Skills 的设计跟公网场景有几个明显区别。第一步骤要更明确。公网模型能力强可以给个模糊指令让它自己发挥。内网模型往往是量化过的能力弱一些Skill 的步骤必须写得非常具体每一步做什么、输入什么、输出什么都要说清楚。第二容错要更强。内网工具可能不稳定Skill 里要设计重试和降级逻辑。比如查数据库失败是重试三次还是返回缓存要在 Skill 里定义清楚。第三输出要结构化。内网 Agent 往往要对接下游系统输出格式必须严格。我一般要求 Skill 的输出是 JSON字段名和类型都固定方便下游解析。下面是一个内网 Skills 的配置示例用 YAML 描述name: employee_query_skill description: 查询员工信息并生成结构化报告 trigger: keywords: [查员工, 员工信息, 工号查询] steps: - name: extract_name description: 从用户输入中提取员工姓名 tool: llm_extract input: {{user_input}} output: {{employee_name}} - name: query_db description: 调用数据库工具查询 tool: mcp__internal-db-server__query_employee input: name: {{employee_name}} retry: 3 fallback: 返回查询失败请稍后重试 output: {{raw_result}} - name: format_output description: 格式化为标准 JSON tool: llm_format input: {{raw_result}} output_schema: type: object properties: name: {type: string} department: {type: string} position: {type: string}这个 Skill 定义了三步提取姓名、查数据库、格式化输出。每一步都指定了工具、输入、输出还有重试和降级。Agent 拿到这个 Skill就知道该怎么干活了。4.3 Skills 的加载与热更新内网环境改代码成本高Skills 最好支持热更新。我的做法是把 Skills 配置放在一个目录里Agent 启动时加载同时起一个文件监听配置变了自动重载。import os import yaml from watchdog.observers import Observer from watchdog.events import FileSystemEventHandler class SkillLoader: def __init__(self, skill_dir): self.skill_dir skill_dir self.skills {} self.load_all() def load_all(self): for filename in os.listdir(self.skill_dir): if filename.endswith(.yaml): path os.path.join(self.skill_dir, filename) with open(path, r, encodingutf-8) as f: skill yaml.safe_load(f) self.skills[skill[name]] skill def get_skill(self, name): return self.skills.get(name) class SkillWatcher(FileSystemEventHandler): def __init__(self, loader): self.loader loader def on_modified(self, event): if event.src_path.endswith(.yaml): self.loader.load_all() print(fSkills reloaded: {event.src_path})这样运维改 Skill 配置不用重启 Agent改完保存就生效。这个机制在内网特别实用因为内网重启服务往往要走审批流程能省一次是一次。4.4 Skills 的测试与灰度Skills 上线前一定要测。内网环境没法像公网那样快速迭代一个 Skill 出问题可能影响整个业务流程。我的做法是给每个 Skill 配一组测试用例包含正常输入、边界输入、异常输入跑通了才允许上线。test_cases [ {input: 查一下张三的信息, expected_name: 张三}, {input: 帮我查李四, expected_name: 李四}, {input: 查一下不存在的员工, expected_error: True}, ] def test_skill(skill, cases): for case in cases: result run_skill(skill, case[input]) if expected_name in case: assert case[expected_name] in result, fFailed: {case} if case.get(expected_error): assert 失败 in result or 错误 in result, fFailed: {case}灰度的话可以给 Skill 加一个enabled字段先在小范围 Agent 实例上开启观察一段时间没问题再全量。这个思路跟公网灰度一样只是内网观察周期要拉长。5. 内网 Agent 的并发扛压实战5.1 并发瓶颈到底在哪很多人一上来就优化模型推理其实内网 Agent 的并发瓶颈往往不在模型而在几个容易被忽略的地方。第一个瓶颈是模型服务的并发上限。vLLM 虽然支持并发但显存有限并发数上去之后请求会排队。你要根据显存和模型大小算出一个合理的并发数超过就限流。第二个瓶颈是MCP 工具调用的阻塞。如果工具是同步的一个请求卡住会拖垮整个 Agent。必须把工具调用改成异步或者用线程池隔离。第三个瓶颈是Agent 自身的状态管理。多轮对话要维护上下文如果上下文存在内存里并发一高内存就爆。要么用 Redis 之类的内网缓存要么限制单实例的会话数。第四个瓶颈是日志和监控。并发高的时候日志写入本身就会成为瓶颈尤其是同步写文件。要用异步日志或者写到内网的消息队列。5.2 用异步和连接池扛并发Python 的 asyncio 是内网 Agent 扛并发的基础。所有 IO 操作都要异步化包括模型调用、工具调用、数据库查询。import asyncio import aiohttp from aiohttp import ClientSession class AgentRuntime: def __init__(self, model_url, mcp_urls): self.model_url model_url self.mcp_urls mcp_urls self.session None self.semaphore asyncio.Semaphore(10) # 限制并发数 async def start(self): self.session ClientSession( timeoutaiohttp.ClientTimeout(total300), connectoraiohttp.TCPConnector(limit50, limit_per_host20) ) async def call_model(self, messages): async with self.semaphore: async with self.session.post( f{self.model_url}/v1/chat/completions, json{model: local-model, messages: messages} ) as resp: return await resp.json() async def call_mcp(self, url, tool_name, args): async with self.session.post( f{url}/call_tool, json{name: tool_name, arguments: args} ) as resp: return await resp.json() async def close(self): await self.session.close()这里有几个关键点Semaphore限制并发数防止把模型服务打爆TCPConnector的连接池复用连接避免频繁建连ClientTimeout设大因为模型推理慢。5.3 请求队列与背压并发再高也有上限超过上限的请求要排队不能直接拒绝也不能无限堆积。我用的是asyncio.Queue加背压机制。class RequestQueue: def __init__(self, max_size100, worker_count5): self.queue asyncio.Queue(maxsizemax_size) self.worker_count worker_count self.workers [] async def submit(self, request): try: self.queue.put_nowait(request) return {status: accepted} except asyncio.QueueFull: return {status: rejected, reason: queue_full} async def worker(self): while True: request await self.queue.get() try: await self.process(request) finally: self.queue.task_done() async def start(self): for _ in range(self.worker_count): self.workers.append(asyncio.create_task(self.worker()))队列满了直接返回“忙”让调用方自己决定重试还是放弃。这比无限堆积导致雪崩要好得多。5.4 压测与容量规划内网上线前一定要压测。我用的是 locust写一个简单的压测脚本from locust import HttpUser, task, between class AgentUser(HttpUser): wait_time between(1, 3) task def chat(self): self.client.post(/v1/chat, json{ message: 查一下张三的信息, session_id: test-session })压测时重点看几个指标P99 延迟、错误率、模型服务的 GPU 利用率、MCP 工具的响应时间。根据压测结果反推容量比如单实例能扛 20 QPS业务峰值是 100 QPS那就至少部署 5 个实例加负载均衡。注意压测要在内网做别拿公网数据估。内网网络延迟、机器性能都跟公网不一样估出来的数不准。6. 依赖离线化把整个工具链搬进内网6.1 Python 依赖的离线打包Python 依赖离线化是内网部署最烦的一环因为依赖树深、平台相关性强。我的标准流程是# 在联网机器上用跟内网一致的环境 pip download -r requirements.txt -d ./packages \ --platform manylinux2014_x86_64 \ --python-version 310 \ --only-binary:all:--only-binary:all:强制只下 wheel 包避免下到源码包在内网编译。如果某个包没有 wheel就得单独处理要么找替代包要么在内网准备好编译环境。拷进内网后安装pip install --no-index --find-links./packages -r requirements.txt如果报缺依赖用pip check查然后回联网机器补下。6.2 模型权重的离线搬运模型权重动辄几个 G 到几十个 G搬运是个体力活。我的做法是用huggingface-cli download在联网机器下好然后用rsync或者移动硬盘拷进内网。# 联网机器下载 huggingface-cli download Qwen/Qwen2.5-7B-Instruct --local-dir ./qwen2.5-7b # 校验完整性 huggingface-cli download Qwen/Qwen2.5-7B-Instruct --local-dir ./qwen2.5-7b --resume拷进去之后用sha256sum校验关键文件确保没损坏。大文件传输损坏是常事校验这一步不能省。6.3 内网自建包仓库如果内网机器多每次都手动拷包太低效。我的做法是在内网起一个私有的 PyPI 仓库用devpi或者pypiserver。# 用 pypiserver 起一个最简单的私有源 pip install pypiserver pypi-server run -p 8080 -a . -P . /data/pypi-packages然后把联网机器下好的包上传进去twine upload --repository-url http://内网IP:8080 --username admin --password admin ./packages/*内网机器配置 pip 源指向这个私有仓库pip config set global.index-url http://内网IP:8080/simple pip config set global.trusted-host 内网IP这样内网机器装包就跟公网一样方便了只是源换成了内网。6.4 容器镜像的离线导入如果内网用容器部署镜像也要离线导入。流程是联网机器docker pull然后docker save成 tar 包拷进内网docker load。# 联网机器 docker pull vllm/vllm-openai:latest docker save vllm/vllm-openai:latest -o vllm.tar # 内网机器 docker load -i vllm.tar镜像大的话docker save出来的 tar 包也大可以用pigz压缩加速传输。另外如果内网有 Harbor 之类的私有镜像仓库直接 push 进去更省事。7. 几个我踩过的坑和对应的解法7.1 模型加载慢到怀疑人生第一次在内网部署 14B 模型加载花了将近二十分钟我以为卡死了。后来发现是磁盘 IO 太慢模型文件放在机械盘上。换成 SSD 之后加载时间降到三分钟以内。解法模型文件一定放 SSD别放机械盘或者网络存储。如果内网只有机械盘考虑用mmap方式加载或者把模型转成更紧凑的格式。7.2 MCP 工具调用超时导致 Agent 卡死有一次 Agent 调用一个内网 API 工具那个 API 挂了Agent 一直等整个会话卡死。后来加了超时和熔断才解决。解法所有 MCP 工具调用必须设超时超时后返回错误而不是一直等。再加一个熔断器某个工具连续失败就暂时禁用过一段时间再试。from circuitbreaker import circuit circuit(failure_threshold5, recovery_timeout60) async def call_tool_with_breaker(url, tool_name, args): return await call_mcp(url, tool_name, args)7.3 Skills 配置写错导致 Agent 乱调工具有个 Skill 的触发关键词写得太宽泛结果用户随便说句话就触发Agent 乱调工具。后来把触发条件收紧加了意图识别才解决。解法Skill 的触发条件要尽量精确能用意图识别就别用关键词匹配。关键词匹配容易误触发尤其是在中文场景下。7.4 并发一高日志就丢内网 Agent 并发上去之后发现日志丢了不少。排查发现是同步写文件高并发下写入冲突。改成异步日志队列之后解决。解法日志用logging.handlers.QueueHandler加QueueListener写入操作放到单独线程主流程只往队列里丢。import logging import logging.handlers import queue log_queue queue.Queue(-1) queue_handler logging.handlers.QueueHandler(log_queue) root logging.getLogger() root.addHandler(queue_handler) file_handler logging.FileHandler(/var/log/agent.log) listener logging.handlers.QueueListener(log_queue, file_handler) listener.start()7.5 内网时间不同步导致 Token 过期内网机器时间不同步导致 JWT Token 校验失败Agent 调工具一直 401。排查了半天才发现是时间问题。解法内网一定要配 NTP 服务所有机器时间同步。这个坑很隐蔽但一旦踩上很难查。8. 内网 Agent 工程的持续演进思路内网 Agent 上线只是开始后面怎么演进才是关键。我分享几个实际用过的方向。第一个方向是能力沉淀。每解决一个新场景就把它固化成一个 Skill慢慢积累成内网的能力库。时间长了Agent 能干的活越来越多价值越来越大。第二个方向是效果评估。内网没法像公网那样快速 A/B 测试但可以建一个内部评测集每次模型或 Skill 更新都跑一遍确保效果不退化。评测集要覆盖核心场景包含正常和异常输入。第三个方向是资源优化。内网资源有限要持续优化。比如模型量化策略调整、推理框架升级、缓存机制引入都能在不加硬件的情况下提升吞吐。第四个方向是可观测性。内网 Agent 出问题排查难要把日志、指标、链路追踪做起来。至少要有请求级别的日志能追溯每个请求经过了哪些步骤、调了哪些工具、耗时多少。第五个方向是安全加固。内网不等于安全权限控制、审计日志、敏感操作二次确认这些都要逐步补上。尤其是 Agent 能操作内部系统之后安全边界要划清楚。我在实际项目里的体会是内网 Agent 工程最难的不是技术而是心态。公网环境习惯了“缺什么装什么”内网环境必须“有什么用什么”。这个转变需要时间但一旦转过来你会发现内网 Agent 反而更可控、更稳定因为所有环节都在自己手里没有外部依赖的不确定性。