1. 为什么要在隔离内网里折腾 AI Agent
先把场景说清楚。所谓隔离内网,就是那种物理上跟公网断开、或者只允许极少数白名单流量进出的网络环境。银行的核心机房、制造业的产线控制网、部分科研院所的算力集群,都属于这一类。在这种环境里谈 AI Agent 工程化,跟在公网上完全是两码事——你在公网能随手pip install的东西,在这里可能连包都传不进去;你在公网能直接调用的云端大模型 API,在这里连 DNS 都解析不了。
我前后在三个不同规模的内网环境里落地过 AI Agent 相关的工程,从最初级的"本地跑个模型 + 写死几个工具函数",到后来引入 MCP 协议做工具编排、用 Skills 做能力封装,踩的坑足够写一本小册子。这篇就把整套思路和实操细节摊开讲,重点放在怎么在没有外网的前提下,把 Agent 的工具调用、能力扩展、并发承载这几件事做扎实。
适合谁看?如果你手上有内网服务器、有本地部署的模型(不管是 DeepSeek 系列、Qwen 系列还是别的开源权重),并且想让 Agent 真正干点活——比如自动处理工单、解析内部文档、调用内部系统的接口——那这篇就是给你写的。不需要你有多深的分布式经验,但得能看懂基本的 Linux 命令和 Python 代码。
核心关键词先摆出来:AI Agent、MCP、Skills、内网、工程实战。这五个词贯穿全文,后面每一节都会围绕它们展开。MCP 是 Model Context Protocol,简单理解就是"让模型和外部工具之间有一套标准对话方式"的协议;Skills 则是把一组相关能力打包成一个可复用的单元。这两个概念在内网环境下尤其重要,因为内网最缺的就是"现成的、能直接用的东西",你得自己造轮子,而 MCP 和 Skills 就是造轮子的模具。
2. 内网 Agent 的整体架构怎么定
2.1 先想清楚:哪些东西必须在内网,哪些可以妥协
架构设计的第一步不是画图,是划边界。我见过太多人一上来就想着"把公网那套搬进来",结果卡在依赖下载上动弹不得。正确的做法是先列一张清单,把组件分成三类:
- 必须在内网:模型推理服务、Agent 运行时、工具执行器、数据存储。这些涉及核心数据和算力,出不去。
- 可以离线搬运:模型权重、Python 依赖包、Docker 镜像。这些通过一次性摆渡导入即可。
- 能省则省:监控、日志聚合这类辅助组件,内网环境下能用轻量方案就别上重型栈。
我一般会画一个三层结构:最底层是模型服务层,跑本地推理;中间是Agent 编排层,负责对话管理、工具调度、上下文维护;最上层是能力层,由一个个 MCP Server 和 Skills 组成,对外暴露具体功能。
2.2 模型服务层:本地推理怎么选型
内网跑模型,绕不开推理框架的选择。目前主流的有几条路线:vLLM、SGLang、llama.cpp、Ollama。我的经验是这样分的:
| 框架 | 适用场景 | 显存要求 | 内网部署难度 |
|---|---|---|---|
| vLLM | 多并发、生产级 | 较高,需 A100/4090 级别 | 中,依赖较多 |
| SGLang | 高吞吐、结构化输出 | 较高 | 中 |
| llama.cpp | 单机、低资源 | 低,CPU 也能跑 | 低 |
| Ollama | 快速验证、小规模 | 中 | 低 |
如果内网有像样的 GPU 资源,vLLM 是首选,它的 PagedAttention 对显存利用率提升明显,并发场景下吞吐能比朴素实现高好几倍。如果只有 CPU 或者显存紧张,llama.cpp 配合量化权重(Q4_K_M 这种)也能跑起来,只是响应速度会慢一些,适合对延迟不敏感的场景。
提示:内网导入模型权重时,优先选 GGUF 或 safetensors 格式,前者适配 llama.cpp,后者适配 vLLM/SGLang。别用那些需要在线转换的格式,内网转换会要命。
2.3 Agent 编排层:为什么我最终选了 MCP 做工具层
早期我做内网 Agent,工具调用是写死在代码里的——一个tools.py文件,里面几十个函数,Agent 通过函数名匹配来调用。这套方案在工具少于 10 个的时候还能用,一旦超过 20 个,维护成本就爆炸了:改一个工具要动主程序,加一个工具要重新测试整个链路。
MCP 的出现解决了这个问题。它的核心思想是把工具的定义和执行从 Agent 主程序里解耦出来,每个 MCP Server 是一个独立进程,通过标准协议(stdio 或 SSE)跟 Agent 通信。Agent 只需要知道"有哪些 Server 可用",具体每个 Server 提供什么工具,运行时动态发现。
这个设计在内网环境下有三个明显好处:
- 独立部署:每个 MCP Server 可以单独更新,不用重启整个 Agent。
- 权限隔离:不同 Server 跑在不同用户下,敏感操作单独管控。
- 语言无关:Server 可以用 Python、Node、Rust 任意语言写,只要实现协议即可。
我现在的标准做法是:Agent 主程序用 Python 写,MCP Server 按功能域拆分,比如mcp-file(文件操作)、mcp-db(数据库查询)、mcp-http(内部接口调用),每个都是独立进程。
2.4 能力层:Skills 到底该怎么切分
Skills 这个概念容易跟 MCP 混淆。我的理解是:MCP 解决的是"工具怎么被调用",Skills 解决的是"一组能力怎么被组织"。一个 Skill 可能包含多个 MCP 工具,加上提示词模板、参数校验逻辑、后处理步骤,打包成一个面向具体任务的单元。
举个例子,"生成周报"这个 Skill,背后可能调用了mcp-db查数据、mcp-file读模板、再经过一段提示词加工,最后输出成文档。用户只需要说"帮我生成这周的周报",Agent 就知道该走哪条链路。
切分 Skills 的原则我总结成一句话:按任务切,不按工具切。一个 Skill 对应一类用户意图,而不是对应一个技术模块。
3. 内网环境下的依赖与资源准备
3.1 离线依赖包的完整搬运流程
这是内网工程最烦人但也最基础的一环。我的标准流程是这样的:
第一步,在公网机器上准备一个跟内网目标机完全一致的环境(Python 版本、系统架构、glibc 版本都要对齐)。这一步偷懒,后面必翻车。
第二步,用pip download把依赖下到本地目录:
pip download -r requirements.txt -d ./offline_packages \ --platform manylinux2014_x86_64 \ --python-version 310 \ --only-binary=:all:注意--platform和--python-version这两个参数,它们决定了下载的 wheel 包适配哪个环境。如果内网是 ARM 架构,这里要改成manylinux2014_aarch64。
第三步,把整个目录打包,通过摆渡介质导入内网,然后:
pip install --no-index --find-links=./offline_packages -r requirements.txt--no-index是关键,它强制 pip 不去联网,只用本地包。
注意:有些包在安装时会触发编译(比如带 C 扩展的),这时候
--only-binary=:all:会失败。遇到这种情况,要么在公网机器上预先编译好 wheel,要么在内网机器上准备好编译工具链。我一般倾向于前者,内网编译环境往往缺东少西。
3.2 Docker 镜像的离线导入
如果内网允许用 Docker,事情会简单很多。流程是:
# 公网机器上 docker pull your-image:tag docker save your-image:tag -o image.tar # 摆渡到内网后 docker load -i image.tar镜像里的依赖已经打包好了,不用担心 pip 的问题。但要注意镜像的基础层也得一起 save,docker save默认会包含所有层,所以 tar 包可能很大,几个 G 是常态。
3.3 模型权重的导入与校验
模型权重动辄几十 G,导入过程容易出错。我的做法是分片传输 + 校验和验证:
# 公网侧生成校验和 sha256sum model-00001-of-00004.safetensors > checksums.txt # 内网侧验证 sha256sum -c checksums.txt校验通过再加载,避免因为传输损坏导致模型加载失败却找不到原因。这个坑我踩过一次,排查了大半天才发现是权重文件少了一个字节。
4. MCP Server 在内网的具体实现
4.1 用 Python 写一个最小可用的 MCP Server
MCP 协议本身不复杂,核心就是 JSON-RPC 消息的收发。官方有 Python SDK,但内网环境下你可能需要自己实现精简版。下面是一个最小可用的 stdio 模式 Server 骨架:
import sys import json def handle_request(req): method = req.get("method") if method == "initialize": return {"protocolVersion": "2024-11-05", "capabilities": {"tools": {}}} elif method == "tools/list": return {"tools": [ { "name": "read_file", "description": "读取内网指定路径的文件", "inputSchema": { "type": "object", "properties": {"path": {"type": "string"}}, "required": ["path"] } } ]} elif method == "tools/call": params = req.get("params", {}) if params.get("name") == "read_file": path = params["arguments"]["path"] with open(path, "r", encoding="utf-8") as f: content = f.read() return {"content": [{"type": "text", "text": content}]} return {"error": "unknown method"} def main(): for line in sys.stdin: line = line.strip() if not line: continue req = json.loads(line) resp = handle_request(req) resp["jsonrpc"] = "2.0" resp["id"] = req.get("id") sys.stdout.write(json.dumps(resp) + "\n") sys.stdout.flush() if __name__ == "__main__": main()这个骨架跑起来就能被 Agent 识别。实际生产里要加上错误处理、日志、超时控制,但核心逻辑就这么点。
4.2 stdio 还是 SSE:内网场景怎么选
MCP 支持两种传输方式:stdio 和 SSE。内网环境下我几乎总是选 stdio,原因有三:
- stdio 不需要网络端口,进程间通过管道通信,天然规避了防火墙问题。
- 生命周期绑定,Agent 启动时拉起 Server,Agent 退出时 Server 也跟着结束,不会留下孤儿进程。
- 权限控制简单,Server 以什么用户跑,就有什么权限,不需要额外的认证层。
SSE 适合 Server 需要独立部署、被多个 Agent 共享的场景。内网里如果确实有这种需求(比如一个数据库查询 Server 被好几个 Agent 用),再考虑 SSE,但要自己加一层简单的 token 认证。
4.3 工具描述怎么写才能让模型用对
这是很多人忽略的细节。工具能不能被正确调用,很大程度上取决于description写得好不好。我的经验是:
- 描述里要包含"什么时候用",而不只是"这个工具做什么"。比如"当用户询问内部系统的订单状态时使用",比"查询订单"要好得多。
- 参数描述要具体,说明格式和取值范围。
path参数要写"绝对路径,以 /data 开头",别只写"文件路径"。 - 避免功能重叠的工具,两个工具如果做的事差不多,模型会随机选,结果不稳定。
我做过一个对比测试,同一批任务,工具描述优化前后,调用准确率从 60% 出头提升到 90% 以上。这个投入产出比非常高。
5. Skills 的工程化封装
5.1 Skill 的目录结构设计
一个 Skill 在内网里我一般组织成这样的目录:
skills/ weekly_report/ skill.yaml # 元信息:名称、描述、触发条件 prompt.md # 提示词模板 tools.json # 依赖的 MCP 工具列表 postprocess.py # 后处理逻辑skill.yaml里定义触发条件,Agent 加载时读取所有 Skill 的元信息,构建一个路由表。用户输入进来后,先做意图匹配,命中哪个 Skill 就走哪条链路。
5.2 提示词模板的参数化
Skill 的提示词不能写死,要支持参数注入。我用的是简单的占位符替换:
你是一个周报生成助手。请根据以下数据生成周报: 时间范围:{{start_date}} 至 {{end_date}} 数据内容: {{data_content}} 要求: 1. 按项目分组 2. 每个项目列出完成事项和待办 3. 语言简洁,不超过 500 字Agent 在执行时把实际数据填进去。这样做的好处是提示词可以独立迭代,不用改代码。
5.3 Skill 之间的组合与复用
复杂任务往往需要多个 Skill 串联。我实现了一个简单的编排器,支持在skill.yaml里声明depends_on,Agent 会按依赖顺序执行。比如"月度总结"依赖"周报生成",就会先跑四次周报,再汇总。
提示:Skill 组合要控制深度,超过三层的嵌套会让调试变得极其痛苦。我一般限制在两层,再复杂就拆成独立流程。
6. 并发承载:内网 Agent 怎么扛住压力
6.1 并发瓶颈到底在哪
很多人一提到并发就想着加机器,但 Agent 系统的瓶颈往往不在模型推理,而在工具调用的串行等待。一个请求进来,模型思考 2 秒,调用工具 3 秒,再思考 2 秒,总共 7 秒。如果工具调用能并行,总时间能压到 4 秒左右。
所以优化的第一优先级是识别可并行的工具调用。MCP 协议本身支持并行请求,Agent 编排层要做的就是分析当前步骤里哪些工具没有依赖关系,一起发出去。
6.2 用异步框架改造 Agent 主循环
同步的 Agent 主循环在高并发下会迅速耗尽线程。我现在的标准做法是用asyncio重写:
import asyncio async def call_tool_async(server, tool_name, args): proc = await asyncio.create_subprocess_exec( "python", server, stdin=asyncio.subprocess.PIPE, stdout=asyncio.subprocess.PIPE ) req = json.dumps({"method": "tools/call", "params": {"name": tool_name, "arguments": args}}) stdout, _ = await proc.communicate(req.encode()) return json.loads(stdout) async def run_agent_step(tools): tasks = [call_tool_async(t["server"], t["name"], t["args"]) for t in tools] return await asyncio.gather(*tasks)这样多个工具调用可以真正并行。实测下来,在 8 核机器上,并发 20 个请求时,异步版本的吞吐是同步版本的 3 倍以上。
6.3 模型推理侧的并发控制
vLLM 本身支持 continuous batching,多个请求会自动合并成一个 batch 处理。但要注意max_num_seqs这个参数,它决定了同时处理多少个序列。设太小浪费吞吐,设太大显存爆掉。我的经验值是显存的 70% 左右留给 KV cache,剩下的给模型权重。
如果内网用的是 llama.cpp,它也有--parallel参数控制并发槽位,但效果不如 vLLM 明显,适合并发不高的场景。
6.4 限流与降级策略
内网资源有限,必须有兜底。我一般加两层保护:
- 入口限流:用信号量控制同时处理的请求数,超过就排队。
- 超时降级:单个工具调用超过 N 秒直接返回失败,让 Agent 走备选路径,而不是一直卡着。
sem = asyncio.Semaphore(10) async def guarded_call(tool): async with sem: try: return await asyncio.wait_for(call_tool_async(**tool), timeout=15) except asyncio.TimeoutError: return {"error": "timeout"}这套组合拳下来,即使突发流量,系统也不会雪崩。
7. 常见问题与排查技巧实录
7.1 问题速查表
| 现象 | 可能原因 | 排查方向 |
|---|---|---|
| Agent 启动后工具列表为空 | MCP Server 未正确响应 initialize | 手动跑 Server,看 stdio 输出 |
| 工具调用返回乱码 | 编码不一致 | 检查 Server 和 Agent 的 locale 设置 |
| 模型不调用工具,直接编答案 | 工具描述不清晰 | 优化 description,加使用场景 |
| 并发一高就卡死 | 同步阻塞 | 检查是否有同步 IO 未改异步 |
| 权重加载报显存不足 | 量化精度太高 | 换更激进的量化,或减小 max_model_len |
| Skill 匹配错误 | 触发条件重叠 | 检查 skill.yaml 的匹配规则 |
7.2 几个我踩过的坑
坑一:stdio 缓冲区死锁。MCP Server 如果输出大量数据但 Agent 没及时读,管道缓冲区满了就会卡住。解决办法是 Server 侧分块输出,Agent 侧持续读取。
坑二:模型对中文工具名支持差。早期我用中文命名工具,发现模型经常调错。改成英文名 + 中文描述后,准确率明显提升。
坑三:内网时间不同步导致日志混乱。多台机器部署时,如果 NTP 没配好,日志时间戳对不上,排查问题极其痛苦。部署前一定先统一时间。
坑四:Skill 的提示词里带了外网链接。内网访问不了,模型会一直尝试,浪费 token。所有提示词里的 URL 都要替换成内网可达的地址,或者直接删掉。
7.3 日志与可观测性
内网没有现成的监控栈,我用的是最土但最有效的办法:结构化日志 + 本地文件轮转。每条日志包含request_id、step、tool、duration、status,出问题时用grep就能串起整个链路。
import logging import json logger = logging.getLogger("agent") def log_step(request_id, step, tool, duration, status): logger.info(json.dumps({ "request_id": request_id, "step": step, "tool": tool, "duration_ms": duration, "status": status }))别小看这个,内网排查问题全靠它。我甚至写了个小脚本,把日志按request_id聚合,直接输出一次完整请求的调用链,效率提升巨大。
8. 一些工程之外的体会
内网做 AI Agent,技术难度其实不是最大的,最大的挑战是信息孤岛。你在公网遇到问题,搜一下就有答案;在内网,只能靠自己啃文档、读源码、做实验。所以我的建议是:在公网阶段就把所有可能用到的资料、文档、示例代码整理好,一次性带进去。我现在的习惯是维护一个"内网知识包",包含常用库的离线文档、MCP 协议的完整规范、几个典型 Skill 的参考实现,每次进内网前更新一遍。
另一个体会是别追求一步到位。我第一个内网 Agent 只有三个工具,能查文件、能跑 SQL、能发内部通知,但就是这三个工具,解决了团队 80% 的重复劳动。后来慢慢加 MCP、加 Skills、加并发优化,都是在这个基础上迭代出来的。先跑起来,再跑好,这个顺序不能反。
最后分享一个实用的小技巧:内网 Agent 的提示词里,我习惯加一句"如果不确定,先调用工具确认,不要凭记忆回答"。这句话能显著降低幻觉率,尤其是在处理内部数据的时候。成本几乎为零,效果立竿见影。