1. 项目缘起与整体架构设计
1.1 为什么要在隔离内网做 AI Agent
先说清楚一件事:隔离内网不是"没有网线的局域网",而是物理隔离、协议隔离、单向导入这类真正意义上的封闭环境。银行核心机房、军工研究所、部分制造业的工控网段,都属于这个范畴。在这些地方做 AI Agent,和你在大厂云上跑一个 LangChain 应用,完全是两码事。
我最初接到这个需求的时候,第一反应是"这活儿能干吗"。因为绝大多数 AI Agent 框架的默认假设是:你能随时访问模型 API、能拉取依赖包、能连外网做工具调用。而隔离内网里,这三条全部不成立。你要解决的不是"怎么让 Agent 更聪明",而是"怎么让 Agent 先跑起来"。
核心矛盾集中在四个层面:
- 模型层:无法调用云端大模型 API,必须本地部署推理服务
- 依赖层:pip、npm、cargo 全部无法直连,所有依赖需要离线搬运
- 工具层:MCP 协议依赖的很多工具需要网络,需要重新设计工具链
- 运维层:没有公网入口,日志、监控、更新全部要走内网通道
这四个矛盾决定了整个工程的设计思路。我的方案是:把 Agent 拆成"推理内核 + 工具总线 + 编排层"三段,每一段都做离线化改造。推理内核用本地部署的开源模型,工具总线用 MCP 协议做标准化封装,编排层用轻量级框架自己写,避免引入过重的依赖。
为什么选 MCP 而不是自己造一套工具协议?因为 MCP 的抽象层次刚好合适——它把"工具描述"和"工具调用"分离,工具端只需要暴露一个标准接口,Agent 端只需要理解这个接口。在内网环境里,这意味着你可以在不同机器上部署不同工具,通过内网 HTTP 或 stdio 通信,而不需要每加一个工具就改 Agent 代码。
1.2 整体架构的分层拆解
整个系统我分成了五层,从下往上依次是:
| 层级 | 职责 | 关键技术选型 | 内网适配要点 |
|---|---|---|---|
| 硬件层 | GPU 算力、存储、网络 | 国产 GPU 或消费级显卡 | 显存决定模型规模上限 |
| 模型层 | 本地推理服务 | vLLM / Ollama / llama.cpp | 离线权重、量化格式 |
| 协议层 | 工具标准化 | MCP 协议 | stdio 优先,HTTP 备选 |
| 编排层 | Agent 逻辑 | 自研轻量框架 | 零外部依赖 |
| 应用层 | 具体业务场景 | Skills 封装 | 按需加载 |
这个分层的好处是:每一层都可以独立替换。比如模型层从 Ollama 换成 vLLM,上层完全无感;协议层从 stdio 换成 HTTP,工具端只需要改一个 transport 配置。
我特别想强调协议层用 stdio 优先这个决策。很多人一上来就想用 HTTP 做 MCP 通信,觉得"网络调用更通用"。但在隔离内网里,stdio 有三个压倒性优势:第一,不需要开端口,不触发安全策略;第二,进程生命周期和 Agent 绑定,不会出现"工具服务挂了但 Agent 不知道"的情况;第三,调试简单,直接看标准输入输出就行。HTTP 只在"工具必须部署在另一台机器"时才用。
1.3 模型选型的取舍逻辑
内网部署模型,第一个要回答的问题是:显存有多少。这直接决定了你能跑多大的模型。
我的经验值是:
- 7B 模型,4-bit 量化,需要约 5-6GB 显存
- 14B 模型,4-bit 量化,需要约 10-12GB 显存
- 32B 模型,4-bit 量化,需要约 20-24GB 显存
- 70B 模型,4-bit 量化,需要约 40-48GB 显存
如果只有一张 24GB 的卡,32B 量化模型是性价比最高的选择。再大就要多卡,而多卡在内网环境里的部署复杂度会陡增。
模型格式我推荐GGUF。原因是 llama.cpp 对 GGUF 的支持最成熟,CPU 推理也能跑,而且量化选项丰富(Q4_K_M、Q5_K_M、Q8_0 等)。如果 GPU 资源充足,vLLM 的吞吐更高,但它对模型格式要求更严格,且依赖较多,离线安装麻烦。
提示:内网部署模型时,一定要提前确认推理框架的 CUDA 版本和显卡驱动版本是否匹配。我踩过一次坑,驱动版本低了两个小版本,vLLM 直接起不来,排查了半天。
2. 离线依赖搬运与 MCP 工具链搭建
2.1 依赖离线化的完整流程
这是整个工程里最枯燥但最不能出错的部分。隔离内网没有 pip 源,所有 Python 包、系统库、模型权重都要靠"摆渡"进去。
我的标准流程是三步:
第一步,在联网机器上构建完整依赖树。
# 创建干净的虚拟环境 python -m venv agent_env source agent_env/bin/activate # 安装所有依赖 pip install -r requirements.txt # 导出完整依赖清单(含间接依赖) pip freeze > full_requirements.txt # 下载所有 wheel 包到本地目录 pip download -r full_requirements.txt -d ./offline_packages \ --platform manylinux2014_x86_64 \ --python-version 310 \ --only-binary=:all:这里有个关键点:--platform和--python-version必须和内网目标机器完全一致。我见过有人在内网机器上 Python 3.10,联网机器上用 3.11 下载包,结果一堆 wheel 装不上。
第二步,处理非 Python 依赖。
MCP 工具经常需要一些系统级依赖,比如libssl、libffi、sqlite3开发库。这些在联网机器上用apt-get download或者直接拷贝.deb包。
# 下载 deb 包及其依赖 apt-get download libssl-dev libffi-dev # 或者用 apt-rdepends 递归下载 apt-rdepends libssl-dev | grep -v "^ " | xargs apt-get download第三步,内网安装与验证。
# 内网机器上从本地目录安装 pip install --no-index --find-links=./offline_packages -r full_requirements.txt # 验证关键包 python -c "import mcp; print(mcp.__version__)"注意:
--no-index参数必须加,否则 pip 会尝试联网,在内网里会卡住很久才报错。
2.2 MCP 协议在内网里的落地方式
MCP 的核心价值是工具描述的标准化。一个 MCP Server 暴露一组工具,每个工具用 JSON Schema 描述输入输出,Agent 端通过标准协议调用。
在内网环境里,我推荐用stdio transport。具体做法是:把每个 MCP Server 写成一个独立的 Python 脚本,Agent 通过子进程启动它,用标准输入输出通信。
# 一个最简单的 MCP Server 示例 import asyncio from mcp.server import Server from mcp.server.stdio import stdio_server from mcp.types import Tool, TextContent app = Server("internal-tools") @app.list_tools() async def list_tools(): return [ Tool( name="query_database", description="查询内网数据库", inputSchema={ "type": "object", "properties": { "sql": {"type": "string", "description": "SQL 查询语句"} }, "required": ["sql"] } ) ] @app.call_tool() async def call_tool(name: str, arguments: dict): if name == "query_database": result = execute_sql(arguments["sql"]) return [TextContent(type="text", text=str(result))] async def main(): async with stdio_server() as (read, write): await app.run(read, write, app.create_initialization_options()) if __name__ == "__main__": asyncio.run(main())这个模式的好处是:工具和 Agent 完全解耦。你可以在内网任何一台机器上部署 MCP Server,只要 Agent 能通过 stdio 或内网 HTTP 访问到它。
如果工具必须跨机器部署,就用 HTTP transport。但要注意,内网 HTTP 也要做基本的鉴权,不能裸奔。我的做法是加一个简单的 token 校验中间件。
2.3 Skills 的设计与加载机制
Skills 是我在这个项目里最满意的一个设计。它的本质是把"提示词 + 工具组合 + 执行流程"打包成一个可复用的单元。
一个 Skill 包含三部分:
- 元数据:名称、描述、触发条件
- 提示词模板:告诉模型这个 Skill 是干什么的
- 工具依赖:这个 Skill 需要哪些 MCP 工具
加载机制我用的是按需加载。Agent 启动时只加载 Skill 的元数据(很轻量),当用户请求匹配到某个 Skill 的触发条件时,才把完整的提示词和工具依赖加载进来。
class Skill: def __init__(self, name, description, trigger_patterns, prompt_template, tools): self.name = name self.description = description self.trigger_patterns = trigger_patterns self.prompt_template = prompt_template self.tools = tools def matches(self, user_input): return any(re.search(p, user_input) for p in self.trigger_patterns) class SkillRegistry: def __init__(self): self.skills = {} def register(self, skill): self.skills[skill.name] = skill def find_matching(self, user_input): return [s for s in self.skills.values() if s.matches(user_input)]这样做的好处是:上下文窗口不会被无关 Skill 占满。我见过有人把所有 Skill 的提示词一股脑塞进 system prompt,结果 32K 上下文里一半是废话,模型反而变笨了。
实操心得:Skill 的触发条件不要写得太宽泛。比如"查询"这个词,几乎什么请求都能匹配上。我一般用"动词 + 名词"的组合模式,比如"查询数据库"、"生成报表"、"分析日志",精确度会高很多。
3. 核心环节实操与参数调优
3.1 本地推理服务的部署与压测
模型部署我用的是 llama.cpp 的 server 模式,原因是它对 GGUF 支持最好,而且 CPU/GPU 混合推理很灵活。
# 启动 llama.cpp server ./llama-server \ -m /models/qwen2.5-32b-instruct-q4_k_m.gguf \ -c 8192 \ -ngl 99 \ --host 127.0.0.1 \ --port 8080 \ -t 8 \ --parallel 4参数解释:
-c 8192:上下文长度 8K。内网场景下,8K 通常够用,再大显存吃不消-ngl 99:所有层都放到 GPU 上。如果显存不够,可以调小这个值,让部分层跑在 CPU 上-t 8:CPU 线程数,一般设为物理核心数--parallel 4:并发请求数。这个值决定了同时能处理几个请求
并发能力是内网 Agent 的关键指标。我实测下来,32B Q4 模型在单张 24GB 卡上,--parallel 4时每个请求的生成速度大约是 15-20 token/s,--parallel 1时能到 40-50 token/s。所以如果对延迟敏感,就降低并发数;如果对吞吐敏感,就提高并发数。
压测我用的是一个简单的 Python 脚本:
import asyncio import aiohttp import time async def send_request(session, prompt): start = time.time() async with session.post( "http://127.0.0.1:8080/completion", json={"prompt": prompt, "max_tokens": 256} ) as resp: result = await resp.json() elapsed = time.time() - start tokens = result.get("tokens_predicted", 0) return elapsed, tokens async def benchmark(concurrency=4, total=20): async with aiohttp.ClientSession() as session: tasks = [] for i in range(total): tasks.append(send_request(session, f"测试请求 {i}")) results = await asyncio.gather(*tasks) avg_time = sum(r[0] for r in results) / len(results) avg_tokens = sum(r[1] for r in results) / len(results) print(f"平均延迟: {avg_time:.2f}s") print(f"平均生成: {avg_tokens:.0f} tokens") print(f"吞吐: {avg_tokens/avg_time:.1f} tokens/s") asyncio.run(benchmark())3.2 Agent 编排层的实现细节
编排层我没有用 LangChain 或 AutoGen,原因是它们依赖太重,离线安装麻烦,而且很多功能在内网里用不上。我自己写了一个约 500 行的轻量框架,核心是一个 ReAct 循环。
class Agent: def __init__(self, llm_client, skill_registry, mcp_clients): self.llm = llm_client self.skills = skill_registry self.mcp = mcp_clients self.max_iterations = 10 async def run(self, user_input): # 1. 匹配 Skill matched_skills = self.skills.find_matching(user_input) # 2. 构建 system prompt system_prompt = self._build_system_prompt(matched_skills) # 3. ReAct 循环 messages = [ {"role": "system", "content": system_prompt}, {"role": "user", "content": user_input} ] for i in range(self.max_iterations): response = await self.llm.chat(messages) # 解析是否有工具调用 tool_call = self._parse_tool_call(response) if not tool_call: return response # 没有工具调用,直接返回 # 执行工具 tool_result = await self._execute_tool(tool_call) # 把结果加回对话 messages.append({"role": "assistant", "content": response}) messages.append({"role": "tool", "content": tool_result}) return "达到最大迭代次数,任务未完成" def _build_system_prompt(self, skills): prompt = "你是一个内网 AI 助手,可以使用以下工具:\n\n" for skill in skills: prompt += f"## {skill.name}\n{skill.description}\n\n" for tool in skill.tools: prompt += f"- {tool['name']}: {tool['description']}\n" prompt += "\n请根据用户请求,决定是否调用工具。" return prompt这个循环的关键是最大迭代次数。我设的是 10,防止模型陷入死循环。实际用下来,大部分任务 3-5 轮就能完成。
3.3 工具调用的参数校验与容错
内网环境里,工具调用失败是常态。数据库可能连不上,文件可能不存在,权限可能不够。所以参数校验和容错必须做扎实。
我的做法是在 MCP Server 端做三层校验:
def validate_and_execute(tool_name, arguments): # 第一层:Schema 校验 schema = get_tool_schema(tool_name) try: jsonschema.validate(arguments, schema) except jsonschema.ValidationError as e: return {"error": f"参数格式错误: {e.message}"} # 第二层:业务规则校验 if tool_name == "query_database": sql = arguments["sql"].upper() if any(kw in sql for kw in ["DROP", "DELETE", "TRUNCATE"]): return {"error": "不允许执行删除类操作"} # 第三层:执行容错 try: result = execute_tool(tool_name, arguments) return {"result": result} except TimeoutError: return {"error": "工具执行超时"} except PermissionError: return {"error": "权限不足"} except Exception as e: return {"error": f"未知错误: {str(e)}"}实操心得:工具返回的错误信息要尽量具体,但不要暴露内部细节。比如"数据库连接失败"比"psycopg2.OperationalError: could not connect to server"更合适,后者可能泄露内网 IP 和端口。
4. 常见问题排查与避坑实录
4.1 模型推理相关的典型问题
问题一:模型加载后显存溢出。
这个最常见。原因是 GGUF 文件的量化格式和实际显存占用有偏差。比如 Q4_K_M 标称 4-bit,但实际占用可能比理论值高 10-15%。
解决方法:先用nvidia-smi看实际显存占用,然后调整-ngl参数。如果 24GB 卡跑 32B Q4 溢出,就把-ngl从 99 降到 80,让部分层跑 CPU。
问题二:生成速度突然变慢。
可能是上下文长度接近上限了。llama.cpp 在上下文快满的时候,KV Cache 的换入换出会拖慢速度。解决方法是设置--context-shift或者主动截断历史。
问题三:模型输出乱码或重复。
通常是 prompt 格式不对。不同模型的 chat template 不一样,Qwen 用<|im_start|>,Llama 用[INST]。用错模板会导致模型"看不懂"输入。
| 问题现象 | 可能原因 | 排查方法 | 解决方案 |
|---|---|---|---|
| 显存溢出 | 量化格式偏差 | nvidia-smi 看占用 | 降低 -ngl |
| 速度变慢 | 上下文接近上限 | 看日志的 context 使用率 | 截断历史或增大 -c |
| 输出乱码 | chat template 错误 | 对比模型文档 | 用正确的模板 |
| 重复输出 | 温度参数过低 | 检查 temperature | 调到 0.7-0.9 |
4.2 MCP 工具链的常见故障
故障一:MCP Server 启动失败。
最常见的原因是 Python 路径问题。Agent 通过子进程启动 MCP Server 时,用的是系统 Python,而不是虚拟环境的 Python。解决方法是在启动命令里写绝对路径:
subprocess.Popen(["/path/to/venv/bin/python", "mcp_server.py"])故障二:工具调用超时。
内网环境里,工具可能因为网络或权限问题卡住。我的做法是给每个工具调用加超时:
async def call_tool_with_timeout(tool_name, arguments, timeout=30): try: return await asyncio.wait_for( mcp_client.call_tool(tool_name, arguments), timeout=timeout ) except asyncio.TimeoutError: return {"error": f"工具 {tool_name} 执行超时"}故障三:stdio 通信乱码。
这个坑我踩过。原因是 MCP Server 里用了print()输出调试信息,污染了 stdio 通道。解决方法是所有调试信息都写到 stderr,stdout 只留给协议通信。
注意:MCP 协议对 stdio 的纯净度要求很高。任何非协议内容写到 stdout 都会导致解析失败。调试时用
sys.stderr.write(),不要用print()。
4.3 内网部署的独家避坑技巧
技巧一:依赖包用 Docker 镜像搬运。
如果内网允许导入 Docker 镜像,这是最省事的方式。在联网机器上构建好镜像,docker save成 tar 包,内网docker load进去。所有依赖、模型、配置都在镜像里,不用担心版本问题。
技巧二:模型权重分片传输。
大模型权重动辄几十 GB,一次性拷贝容易失败。用split命令分片:
split -b 2G model.gguf model_part_ # 内网合并 cat model_part_* > model.gguf技巧三:日志走文件轮转,不要走 stdout。
内网环境里,Agent 可能跑好几天。日志如果一直往 stdout 写,会把终端缓冲区撑爆。用 Python 的logging.handlers.RotatingFileHandler,按大小轮转。
技巧四:预留降级方案。
如果 GPU 挂了,Agent 要能自动降级到 CPU 推理。虽然慢,但至少能用。我的做法是在配置里写两个模型路径,GPU 版和 CPU 版,启动时先试 GPU,失败就切 CPU。
def load_model(): try: return load_gpu_model("/models/model-gpu.gguf") except Exception as e: logger.warning(f"GPU 模型加载失败: {e},降级到 CPU") return load_cpu_model("/models/model-cpu.gguf")5. 性能优化与扩展方向
5.1 推理性能的进一步压榨
如果对延迟有极致要求,可以上vLLM + PagedAttention。它的核心优势是 KV Cache 的分页管理,显存利用率比 llama.cpp 高不少。但代价是部署复杂度上升,而且对模型格式有要求(需要 HuggingFace 格式,不是 GGUF)。
另一个优化方向是投机采样。用一个小的 draft 模型(比如 1B)先猜几个 token,然后用大模型验证。实测能提升 1.5-2 倍速度。llama.cpp 已经支持这个功能:
./llama-server \ -m /models/qwen2.5-32b-q4.gguf \ -md /models/qwen2.5-1b-q4.gguf \ --draft-max 8 \ --draft-min 45.2 Skills 生态的扩展思路
Skills 做多了之后,管理是个问题。我的做法是建一个Skills 索引,每个 Skill 用 YAML 描述,启动时扫描目录自动加载。
# skills/query_database.yaml name: query_database description: 查询内网数据库并生成报表 trigger_patterns: - "查询.*数据库" - "生成.*报表" tools: - query_database - generate_report prompt_template: | 你是一个数据库查询助手。用户会给你一个查询需求, 你需要先调用 query_database 工具获取数据, 然后调用 generate_report 工具生成报表。这样加新 Skill 只需要加一个 YAML 文件,不用改代码。
5.3 内网 Agent 的监控方案
内网没有 Prometheus 和 Grafana,监控要自己搭。我的方案是用一个简单的 HTTP 端点暴露指标:
from fastapi import FastAPI import psutil app = FastAPI() metrics = {"requests": 0, "errors": 0, "total_tokens": 0} @app.get("/metrics") def get_metrics(): return { **metrics, "cpu_percent": psutil.cpu_percent(), "memory_percent": psutil.virtual_memory().percent, "gpu_memory": get_gpu_memory() }然后用一个内网的小脚本定时拉取,写到日志文件里。虽然简陋,但够用。
我在实际部署中发现,最容易被忽视的是磁盘 I/O。模型加载、日志写入、工具执行都可能产生大量磁盘操作。如果内网机器用的是机械硬盘,模型加载可能要几分钟。换成 SSD 之后,加载时间降到 20 秒以内。这个细节在方案设计阶段就要考虑进去,不然后期迁移很麻烦。
最后再分享一个小技巧:内网 Agent 的配置文件不要硬编码在代码里,用一个单独的config.yaml,这样换环境的时候只改配置就行。我见过有人把模型路径、数据库连接串全写在代码里,换一台机器就要重新改代码、重新打包,效率极低。