1. 为什么要在隔离内网里折腾 AI Agent
第一次接到"隔离内网部署 AI Agent"这个需求时,我的第一反应是:这不是自找麻烦吗?外网环境下一行pip install就能搞定的事,放到隔离内网里,光是依赖包搬运就能耗掉一整天。但真正做过几个项目之后,我的看法完全变了——隔离内网反而是检验一个 AI Agent 工程是否"扎实"的最佳试炼场。
所谓隔离内网,就是物理或逻辑上与公网完全断开的网络环境。金融、能源、制造、医疗这些行业的研发和生产网络,大量采用这种架构。它的核心诉求很简单:数据不出网、外部不可达。但问题也随之而来——现在主流的 AI Agent 方案,几乎都默认你能随时访问模型 API、拉取依赖、调用云端工具。一旦断网,整套链路直接瘫痪。
这篇文章要聊的,就是在这种"要啥没啥"的环境里,怎么把 AI Agent 从零搭起来并跑通。涉及的核心技术点包括MCP(Model Context Protocol)作为工具调用协议、Skills作为能力封装单元、SQLite作为本地持久化方案,以及Vue作为前端交互层。适合正在做内网 AI 落地、或者准备把 Agent 从 Demo 推向生产环境的同学参考。如果你只是想在本地玩个玩具项目,这篇内容可能有点"重",但如果你面对的是真实的内网交付场景,下面这些经验应该能帮你少走不少弯路。
先说一个反直觉的结论:隔离内网下 AI Agent 的最大难点不是模型本身,而是"工具链的离线化"。模型可以量化、可以本地部署,但 Agent 要真正干活,必须能调用工具、读写数据、和前端交互——这些环节每一个都依赖外部资源,每一个都需要你提前做好离线方案。下面我按实际搭建顺序,把整个工程拆开讲。
2. 离线环境下的技术选型:为什么是这套组合
2.1 MCP 协议在内网场景的独特价值
MCP 是 Anthropic 推出的开放协议,本质上是给 AI Agent 和外部工具之间定义了一套标准通信接口。你可以把它理解成"AI 世界的 USB-C"——不管对面是数据库、文件系统还是某个内部系统,只要实现了 MCP Server,Agent 就能用统一的方式调用。
在内网环境里,MCP 的价值比公网更大。原因有三点:
第一,解耦。内网系统往往年代久远、接口五花八门。如果让 Agent 直接对接每个系统的私有 API,代码会变成一团乱麻。MCP 把这些差异封装在 Server 层,Agent 侧只需要认协议,不需要认具体实现。我做过一个项目,内网里有 Oracle、有老式 SOAP 接口、还有几个只能走文件交换的系统,全部包成 MCP Server 之后,Agent 侧的代码量减少了大概 60%。
第二,可控。MCP Server 是你自己写的,跑在内网机器上,调用什么、返回什么、记录什么日志,完全可控。这在合规审计场景下非常关键——每一次工具调用都能留下完整轨迹。
第三,可替换。模型可以换、Agent 框架可以换,只要 MCP 协议不变,工具层不用动。我见过太多项目因为模型升级导致工具调用代码全部重写,用 MCP 就能避免这个问题。
2.2 Skills 的封装粒度怎么定
Skills 这个概念在不同框架里定义不太一样,我这里指的是"Agent 可调用的能力单元"。一个 Skill 可以是一个 MCP 工具,也可以是一段预设的提示词模板,或者两者的组合。
封装粒度是很多人踩坑的地方。我的经验是:按"业务动作"而不是"技术接口"来划分。举个例子,内网里有个查询设备状态的需求。技术接口可能是"调用 HTTP GET /device/status",但业务动作是"查某台设备现在是否正常"。前者是接口粒度,后者是业务粒度。按业务粒度封装,Agent 调用起来更自然,提示词也更好写。
具体做法上,我一般会把 Skills 分成三层:
- 原子层:最基础的操作,比如读文件、查数据库、发 HTTP 请求。这一层尽量通用,不带业务逻辑。
- 组合层:把原子操作串起来完成一个完整动作,比如"根据设备 ID 查询最近 24 小时告警并汇总"。
- 场景层:面向具体业务场景的 Skill,比如"生成设备巡检报告"。
三层分开的好处是复用率高。原子层几乎不用改,组合层偶尔调整,场景层随业务变化。实测下来,一个中等规模的内网 Agent 项目,原子层大概 10-15 个,组合层 20-30 个,场景层 5-10 个就够了。
2.3 SQLite 作为本地存储的取舍
内网环境里,你不太可能给 Agent 配一个独立的数据库集群。SQLite 几乎是唯一合理的选择——零配置、单文件、无需服务进程。
但 SQLite 也有它的脾气。最典型的问题是并发写入。SQLite 默认是库级锁,同一时刻只能有一个写操作。如果你的 Agent 有多个并发任务同时写日志或状态,很容易遇到database is locked错误。
我的处理方案是:
import sqlite3 conn = sqlite3.connect('agent.db', timeout=30, isolation_level=None) conn.execute('PRAGMA journal_mode=WAL') conn.execute('PRAGMA synchronous=NORMAL') conn.execute('PRAGMA busy_timeout=30000')WAL 模式(Write-Ahead Logging)允许读写并发,是内网 Agent 场景下的必开选项。busy_timeout设成 30 秒,给锁等待留足时间。synchronous=NORMAL在 WAL 模式下是安全的,性能比 FULL 好很多。
还有一个容易被忽略的点:十万条数据量级的查询性能。很多人觉得 SQLite 只能处理小数据,其实不然。我实测过,一张 10 万行、20 个字段的表,建好索引之后,带条件的查询基本在 10-50 毫秒之间。关键是要建对索引,而且要用EXPLAIN QUERY PLAN确认索引真的被用上了。
2.4 Vue 前端在内网部署的注意事项
前端这块,Vue 是内网项目里比较稳妥的选择。生态成熟、文档齐全、打包产物是纯静态文件,扔到内网 Nginx 上就能跑。
但内网部署有几个坑必须提前避开:
依赖安装。npm install在内网是跑不通的。解决方案是在外网机器上把node_modules完整打包,或者用npm pack把依赖打成 tarball 带进去。更稳妥的做法是用pnpm的离线模式,把所有依赖缓存到一个目录,内网直接从这个目录安装。
tsconfig 报错。热词里提到的failed to load tsconfig '@vue/tsconfig/tsconfig.web.json'这个错误,本质上是@vue/tsconfig这个包没装上。内网环境下手动把这个包放进node_modules就行,或者干脆在tsconfig.json里不继承它,自己写完整的配置。
构建产物路径。内网部署经常遇到静态资源 404,多半是publicPath没配对。如果前端不是部署在根路径下,一定要在vue.config.js或vite.config.js里设置正确的base。
3. 从零搭建:内网 AI Agent 的完整落地路径
3.1 环境准备:把"外网能做的事"提前做完
内网搭建的第一原则:所有需要联网的操作,都在外网阶段完成。我一般会准备一个"离线包",包含以下内容:
| 类别 | 具体内容 | 备注 |
|---|---|---|
| 运行时 | Python 3.10+、Node.js 18+ | 用官方离线安装包 |
| Python 依赖 | 所有 pip 包及其依赖 | pip download -d ./pkgs -r requirements.txt |
| Node 依赖 | 所有 npm 包 | pnpm fetch或打包 node_modules |
| 模型文件 | 量化后的模型权重 | 根据显存选择量化等级 |
| 数据库 | SQLite 及初始化脚本 | 单文件,直接拷贝 |
| 工具链 | MCP Server 代码及依赖 | 提前测试通过 |
这里有个细节值得说:pip download默认只下载当前平台的 wheel。如果你的外网机器和内网机器架构不同(比如外网是 x86,内网是 ARM),必须加--platform和--python-version参数。我踩过这个坑,搬进去的包全部装不上,只能重新下载。
3.2 MCP Server 的编写与调试
MCP Server 本质是一个遵循 MCP 协议的服务进程。官方提供了 Python 和 TypeScript 的 SDK,内网环境下推荐用 Python 版本,依赖少、部署简单。
一个最小的 MCP Server 长这样:
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_device", description="根据设备ID查询设备状态", inputSchema={ "type": "object", "properties": { "device_id": {"type": "string", "description": "设备唯一标识"} }, "required": ["device_id"] } ) ] @app.call_tool() async def call_tool(name: str, arguments: dict): if name == "query_device": device_id = arguments["device_id"] result = query_device_from_db(device_id) return [TextContent(type="text", text=result)] async def main(): async with stdio_server() as (read, write): await app.run(read, write, app.create_initialization_options())调试阶段有个技巧:先用 stdio 模式跑通,再考虑换成 SSE 或 HTTP。stdio 模式最简单,Agent 直接以子进程方式启动 Server,不需要处理网络和端口。等逻辑稳定了,再改成网络模式支持多客户端。
调试时我习惯写一个简单的测试脚本,直接调用call_tool函数,不经过 Agent。这样能把工具逻辑和 Agent 逻辑分开验证,出问题时定位更快。
3.3 Skills 的注册与调用链路
Skills 注册的核心是"让 Agent 知道有哪些能力可用"。在 MCP 体系下,这通过list_tools自动完成。但实际项目里,光有工具列表不够,还需要给每个 Skill 配上使用说明和示例。
我的做法是在 MCP Server 的description字段里写清楚三件事:这个工具做什么、什么时候用、参数怎么填。描述写得好,Agent 的调用准确率能提升一大截。举个例子:
description: "查询指定设备在指定时间范围内的告警记录。 适用场景:用户询问某设备是否异常、需要查看历史告警时使用。 不适用:查询设备实时状态请用 query_device_status。 参数:device_id 必填,start_time 和 end_time 为 ISO 8601 格式,默认查询最近 24 小时。"这种描述方式看起来啰嗦,但实测下来,Agent 选错工具的概率明显降低。尤其是当你有多个功能相近的工具时,明确的"不适用"说明非常关键。
调用链路上,Agent 的流程是:接收用户输入 → 判断需要哪些 Skill → 调用 MCP 工具 → 处理返回结果 → 生成回复。中间任何一环出问题,都会导致最终结果不对。所以我在每个环节都加了日志,方便排查。
3.4 SQLite 表结构设计与索引优化
Agent 场景下的 SQLite 表设计,和传统业务系统不太一样。核心表一般有这么几张:
- 会话表:记录每次对话的上下文,字段包括 session_id、user_input、agent_response、timestamp。
- 工具调用表:记录每次 Skill 调用,字段包括 call_id、session_id、tool_name、arguments、result、duration。
- 状态表:记录 Agent 的运行时状态,比如当前任务、进度、中间结果。
索引方面,session_id和timestamp是必建的。如果经常按工具名统计,tool_name也要建索引。建索引之前先用EXPLAIN QUERY PLAN看看查询计划,避免建了用不上的索引。
CREATE TABLE tool_calls ( call_id INTEGER PRIMARY KEY AUTOINCREMENT, session_id TEXT NOT NULL, tool_name TEXT NOT NULL, arguments TEXT, result TEXT, duration_ms INTEGER, created_at DATETIME DEFAULT CURRENT_TIMESTAMP ); CREATE INDEX idx_session ON tool_calls(session_id); CREATE INDEX idx_tool_time ON tool_calls(tool_name, created_at);十万条数据量级下,这套结构查询性能完全够用。如果数据量再大,可以考虑按时间分表,或者定期归档历史数据。
4. 踩坑实录:那些文档里不会写的问题
4.1 依赖搬运的"隐性依赖"陷阱
第一次搬依赖时,我以为pip download把requirements.txt里的包下全就完事了。结果内网安装时报了一堆ModuleNotFoundError。排查后发现,很多包有"隐性依赖"——它们不在requirements.txt里,但运行时确实需要。
典型的有:pydantic的 C 扩展、cryptography的底层库、某些包依赖的系统级.so文件。这些在外网环境下会自动装好,但离线搬运时容易漏掉。
我的解决方案是:在外网用pip freeze导出完整依赖树,而不是只导出顶层依赖。然后在干净的环境里测试安装,确认没有遗漏。系统级依赖用ldd检查每个.so文件的依赖,确保内网机器上都有。
4.2 MCP 工具调用的超时与重试
内网环境虽然网络稳定,但工具调用超时还是会发生。原因可能是数据库锁、外部系统响应慢、或者 Agent 本身负载高。
默认的超时设置往往不够用。我的经验是:读操作超时设 30 秒,写操作设 60 秒,涉及外部系统的设 120 秒。重试策略上,读操作可以重试 3 次,写操作要谨慎——重试可能导致重复写入,必须配合幂等设计。
import asyncio from tenacity import retry, stop_after_attempt, wait_exponential @retry(stop=stop_after_attempt(3), wait=wait_exponential(multiplier=1, max=10)) async def call_tool_with_retry(tool_name, arguments): try: return await asyncio.wait_for( mcp_client.call_tool(tool_name, arguments), timeout=30 ) except asyncio.TimeoutError: logger.warning(f"Tool {tool_name} timeout, retrying...") raise这里用tenacity做重试,指数退避避免雪崩。注意wait_exponential的max参数,防止等待时间无限增长。
4.3 Vue 前端与 Agent 的流式交互
Agent 生成回复往往比较慢,如果等全部生成完再返回,用户体验很差。流式输出是必须的。
后端用 SSE(Server-Sent Events)推送,前端用EventSource接收。但内网环境下有几个坑:
Nginx 缓冲。Nginx 默认会缓冲响应,导致 SSE 消息不能实时到达。必须在配置里关掉:
location /api/stream { proxy_pass http://backend; proxy_buffering off; proxy_cache off; proxy_set_header Connection ''; proxy_http_version 1.1; chunked_transfer_encoding off; }前端重连。EventSource断线后会自动重连,但重连时会丢失上下文。我的做法是在 URL 里带上last_event_id,后端根据这个 ID 补发丢失的消息。
Vue 组件卸载。组件销毁时一定要关闭EventSource,否则会内存泄漏。在onUnmounted里调用eventSource.close()。
4.4 模型输出的稳定性控制
内网部署的模型,往往是量化版本,输出稳定性比云端 API 差一些。常见问题包括:格式不按预期、工具调用参数错误、重复输出。
控制手段有几个:
温度调低。Agent 场景下温度设 0.1-0.3 比较合适,太高会导致输出发散。
输出格式约束。在提示词里明确要求 JSON 格式,并给出示例。如果模型支持,开启 JSON mode。
后处理校验。对模型输出做格式校验,不符合的重试或降级处理。我一般会写一个validate_output函数,检查关键字段是否存在、类型是否正确。
工具调用参数校验。MCP 工具的inputSchema会做基础校验,但业务层面的校验还要自己加。比如设备 ID 的格式、时间范围是否合理。
5. 性能调优与稳定性保障
5.1 SQLite 查询优化的实战技巧
前面提到十万条数据量级下 SQLite 性能没问题,但前提是查询写得好。几个实战技巧:
避免SELECT *。只查需要的字段,减少 IO。尤其是表里有大文本字段时,全查会明显变慢。
用EXPLAIN QUERY PLAN验证索引。写完查询后跑一下,看是否走了索引。如果显示SCAN TABLE,说明全表扫描了,需要调整。
批量操作用事务。单条插入一万次和事务里插入一万次,性能差几十倍。
with conn: conn.executemany( "INSERT INTO tool_calls (session_id, tool_name, arguments) VALUES (?, ?, ?)", batch_data )定期VACUUM。删除大量数据后,数据库文件不会自动缩小,需要手动VACUUM。但VACUUM会锁库,建议在低峰期做。
5.2 Agent 并发任务的处理策略
内网 Agent 经常要同时处理多个任务。并发策略上,我推荐任务队列 + 有限并发的模式。
用asyncio.Queue做任务队列,起固定数量的 worker 消费。worker 数量根据机器配置定,一般 CPU 核数的 2-4 倍。每个 worker 处理一个任务,任务内部再并发调用 MCP 工具。
async def worker(queue, worker_id): while True: task = await queue.get() try: await process_task(task) except Exception as e: logger.error(f"Worker {worker_id} error: {e}") finally: queue.task_done() async def main(): queue = asyncio.Queue(maxsize=100) workers = [asyncio.create_task(worker(queue, i)) for i in range(8)] # 投递任务... await queue.join()maxsize设成 100 是防止任务堆积过多导致内存爆掉。如果任务来得太快,投递方会被阻塞,形成背压。
5.3 日志与可观测性建设
内网环境没有云监控,日志就是唯一的可观测手段。我的做法是:
结构化日志。用 JSON 格式输出,方便后续分析。关键字段包括 timestamp、level、session_id、tool_name、duration、error。
分级存储。INFO 级别日志按天切分,保留 30 天。ERROR 级别单独存,保留更久。
关键指标埋点。工具调用成功率、平均耗时、模型响应时间,这些指标定期统计,能提前发现性能退化。
import logging import json class JsonFormatter(logging.Formatter): def format(self, record): log_data = { "timestamp": self.formatTime(record), "level": record.levelname, "message": record.getMessage(), "session_id": getattr(record, "session_id", None), "tool_name": getattr(record, "tool_name", None), } return json.dumps(log_data, ensure_ascii=False)这套日志方案在内网项目里非常实用,出问题时能快速定位到具体环节。
6. 一些个人体会
做内网 AI Agent 这几年,最大的感受是:工程能力比模型能力更重要。模型再强,如果工具链跑不通、数据存不下、前端连不上,整个系统就是空中楼阁。
另一个体会是提前规划离线方案。很多问题在外网环境下根本不会遇到,但内网里会集中爆发。依赖搬运、模型部署、工具调试,每一个环节都要有 Plan B。
最后说一个细节:内网项目的文档一定要写详细。因为内网环境没法随时上网查资料,遇到问题只能靠文档。我现在的习惯是,每解决一个坑,就立刻记下来,包括现象、原因、解决方案。这些积累在后续项目里能省大量时间。
这套方案不是唯一的,但在我做过的几个项目里都跑通了。如果你正在做类似的事情,希望这些经验能帮你少踩几个坑。