☰
隔离内网AI Agent工程实战:MCP与Skills工具调用及并发优化
2026/10/6 5:21:50 网站建设 项目流程

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 提供什么工具,运行时动态发现。

这个设计在内网环境下有三个明显好处:

  1. 独立部署:每个 MCP Server 可以单独更新,不用重启整个 Agent。
  2. 权限隔离:不同 Server 跑在不同用户下,敏感操作单独管控。
  3. 语言无关: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 的提示词里,我习惯加一句"如果不确定,先调用工具确认,不要凭记忆回答"。这句话能显著降低幻觉率,尤其是在处理内部数据的时候。成本几乎为零,效果立竿见影。

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

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

立即咨询