隔离内网AI Agent实战：MCP协议+SQLite+Skills框架离线部署

1. 项目缘起与整体架构设计

1.1 为什么要在隔离内网里折腾 AI Agent

先说清楚这个项目的背景。我所在的团队负责一套工业质检系统的运维和二次开发，生产环境是物理隔离的内网，没有外网出口，连 pip 装包都得走内部镜像源。但业务方看到外面 AI Agent 玩得风生水起，提了个需求：能不能在内网里搞一个能自动查数据库、生成报表、回答运维问题的智能助手。

这个需求听起来简单，实际落地时踩的坑比想象中多得多。外网环境下你随手pip install一个 agent 框架，调个云端大模型 API 就完事了。但内网里，模型要本地部署、工具链要离线打包、依赖要手动搬运，每一步都是体力活加脑力活。

我最终选定的方案是：本地部署开源大模型 + MCP 协议做工具调用 + SQLite 做本地知识库和状态存储 + 自研 Skills 框架做能力扩展。整套系统跑在一台 32GB 内存、带一张 24GB 显存显卡的工控机上，完全离线运行。

这套方案能做什么？简单说，运维人员用自然语言问"上周三产线A的次品率是多少"，Agent 会自动生成 SQL 查询本地 SQLite 数据库，拿到结果后格式化成报表返回。它还能读取本地文档、执行预定义的运维脚本、根据历史工单推荐解决方案。

适合谁来参考？如果你也面临内网环境、数据不能出本地、但又想用上 Agent 能力的场景，这篇内容应该能帮你少走至少两周弯路。如果你只是在外网玩玩 Agent，那这篇的很多坑你可能遇不到，但工具链设计和 Skills 框架的思路同样有参考价值。

1.2 整体架构的分层设计

整套系统我分成了四层，从下往上依次是：

基础设施层负责模型推理和存储。模型用的是本地部署的开源模型，通过兼容 OpenAI 格式的本地推理服务暴露接口。SQLite 承担两个角色：一是业务数据的查询目标，二是 Agent 自身的会话状态、工具调用记录、Skills 注册表的存储。

协议层是 MCP（Model Context Protocol）。这是整个架构的关键。MCP 本质上是一套标准化的工具调用协议，它把"模型想调用某个工具"和"工具实际执行"解耦开。模型只需要输出结构化的调用请求，MCP Server 负责实际执行并返回结果。这样做的好处是，我可以在不重新训练模型的前提下，通过注册新的 MCP Server 来扩展 Agent 的能力。

能力层是 Skills 框架。MCP 解决的是"怎么调工具"，Skills 解决的是"什么时候调、按什么流程调"。一个 Skill 可以理解为一段封装好的业务逻辑，它可能内部调用了多个 MCP 工具，也可能包含条件判断和循环。比如"生成周报"这个 Skill，内部会依次调用"查询本周数据""对比上周数据""生成图表""填充模板"四个步骤。

交互层是前端界面和 API 服务。前端提供一个聊天窗口，API 服务负责接收请求、调度 Agent、返回流式结果。

这里有个设计决策值得说明：我一开始想把 Skills 逻辑直接写进 MCP Server 里，后来发现不行。MCP Server 应该是无状态的、单一职责的，一个 Server 只干一件事。Skills 作为有状态的业务流程编排，必须独立出来。这个分离让后续维护轻松很多。

1.3 技术选型的取舍逻辑

选 SQLite 而不是其他数据库，原因很直接：内网环境没有独立的数据库服务器，SQLite 单文件、零配置、支持标准 SQL，对于十万条级别的数据查询完全够用。实测下来，十万条数据带索引的查询在 50ms 以内，完全满足交互式问答的响应要求。

选 MCP 而不是自己定义一套工具调用格式，是因为 MCP 已经有成熟的生态。虽然内网用不了外网的 MCP Server，但协议本身是开放的，我可以自己实现 Server 端，同时保留未来接入更多工具的可能性。而且 MCP 的流式输出设计很适合 Agent 场景，工具执行过程中的中间结果可以实时推送给前端。

模型选型上，我试过好几个开源模型。最终选择的依据不是跑分，而是工具调用的准确率。有些模型聊天很流畅，但让它输出结构化的工具调用请求就经常格式错误。这个后面会详细说。

2. 核心细节解析与实操要点

2.1 MCP 协议在内网环境的落地要点

MCP 的核心概念其实不复杂，用生活化的类比：它就像给模型配了一个"万能遥控器"。模型不需要知道电视怎么换台，只需要按遥控器上的"换台"按钮。MCP Server 就是那个接收按钮信号并实际执行换台动作的装置。

在内网落地 MCP，第一个要解决的问题是传输方式。外网常用的 SSE（Server-Sent Events）传输在内网会有防火墙和代理的干扰，我最终用的是 stdio 传输——MCP Server 作为子进程启动，通过标准输入输出和主进程通信。这种方式最简单、最稳定，不涉及任何网络端口。

第二个问题是工具描述的编写。MCP 要求每个工具提供 JSON Schema 格式的参数描述。这个描述的质量直接决定模型能不能正确调用。我踩过的坑是：描述写得太简略，模型不知道参数该填什么；写得太复杂，模型又容易理解偏差。

举个例子，查询数据库的工具，我最初的描述是：

{ "name": "query_database", "description": "查询数据库", "parameters": { "type": "object", "properties": { "sql": {"type": "string", "description": "SQL语句"} } } }

结果模型经常生成SELECT * FROM 表名这种不带条件的查询，十万条数据直接拉出来把上下文撑爆。后来我改成：

{ "name": "query_database", "description": "执行只读SQL查询。必须包含WHERE条件限制返回行数，单次查询最多返回100行。禁止使用SELECT *，必须明确指定需要的列名。", "parameters": { "type": "object", "properties": { "sql": { "type": "string", "description": "标准SQLite查询语句，必须包含LIMIT子句，LIMIT值不超过100" }, "purpose": { "type": "string", "description": "本次查询的业务目的，用于审计日志" } }, "required": ["sql", "purpose"] } }

加了约束后，模型生成的 SQL 规范多了。这里的关键经验是：把工具描述当成给新员工的操作手册来写，明确告诉它什么能做、什么不能做、边界在哪里。

2.2 Skills 框架的设计与注册机制

Skills 和 MCP 工具的关系，我打个比方：MCP 工具是"螺丝刀、扳手、锤子"，Skills 是"组装宜家家具的说明书"。说明书告诉你先拧哪个螺丝、再装哪块板，工具只是执行手段。

我的 Skills 框架设计得很轻量，一个 Skill 就是一个 Python 类，继承自基类，实现execute方法。注册机制用的是装饰器模式：

@register_skill( name="weekly_report", description="生成指定产线指定周的质检周报", parameters={ "line_name": {"type": "string", "description": "产线名称"}, "week_offset": {"type": "integer", "description": "周偏移，0表示本周，-1表示上周"} } ) class WeeklyReportSkill(BaseSkill): def execute(self, line_name, week_offset=0): # 第一步：查询本周数据 current_data = self.call_tool("query_database", sql=f"SELECT ... WHERE line='{line_name}' AND week={week_offset}") # 第二步：查询上周数据做对比 previous_data = self.call_tool("query_database", sql=f"SELECT ... WHERE line='{line_name}' AND week={week_offset-1}") # 第三步：生成对比分析 analysis = self.analyze(current_data, previous_data) # 第四步：填充模板 return self.render_template("weekly_report.md", analysis)

这个设计有几个考量。第一，参数用 JSON Schema 描述，和 MCP 工具保持一致，模型理解成本低。第二，Skill 内部可以调用多个 MCP 工具，实现复杂流程编排。第三，Skill 注册表存在 SQLite 里，启动时自动加载，支持热更新。

实操心得：Skills 的粒度要控制好。太细了，模型要调用很多次才能完成一个任务，容易在中间步骤出错；太粗了，灵活性不够。我的经验是，一个 Skill 对应一个完整的业务动作，比如"生成周报""查询设备状态""推荐故障处理方案"，而不是"查询数据""格式化输出"这种技术动作。

2.3 SQLite 作为 Agent 状态存储的实践

SQLite 在这个项目里承担了三个角色，每个角色的表设计都有讲究。

角色一：业务数据查询目标。这是最直接的用途，Agent 生成的 SQL 直接查业务表。这里要注意的是索引设计。十万条数据的表，如果查询字段没索引，全表扫描要几百毫秒；加了索引后降到几毫秒。我的做法是，根据 Agent 最常查询的字段组合建立复合索引。

角色二：会话状态存储。Agent 的多轮对话需要记住上下文。我设计了一张conversation_history表，字段包括会话ID、轮次、角色、内容、时间戳。这里有个坑：上下文不能无限增长，否则会超出模型的上下文窗口。我的策略是保留最近10轮完整对话，更早的对话做摘要压缩后存储。

角色三：工具调用审计日志。每次 MCP 工具调用都记录到tool_call_log表，包括调用的工具名、参数、返回结果、耗时、是否成功。这张表后来成了排查问题的利器——当 Agent 回答错误时，我可以回溯它到底调了什么工具、拿到了什么数据。

CREATE TABLE tool_call_log ( id INTEGER PRIMARY KEY AUTOINCREMENT, session_id TEXT NOT NULL, tool_name TEXT NOT NULL, parameters TEXT, result_summary TEXT, duration_ms INTEGER, success INTEGER, created_at DATETIME DEFAULT CURRENT_TIMESTAMP ); CREATE INDEX idx_session ON tool_call_log(session_id); CREATE INDEX idx_created ON tool_call_log(created_at);

注意：SQLite 默认的并发写入能力有限，多个 Agent 会话同时写入日志时可能遇到database is locked错误。我的解决方案是开启 WAL 模式（PRAGMA journal_mode=WAL），写入性能提升明显，而且读操作不会被写操作阻塞。

2.4 本地模型工具调用能力的调优

这是整个项目最耗时的部分。本地部署的开源模型，聊天能力都不差，但工具调用的准确率参差不齐。我测试了多个模型，工具调用格式错误率从 5% 到 40% 不等。

错误主要分三类：一是格式错误，模型输出的 JSON 不合法，比如多了逗号、少了引号；二是参数错误，JSON 格式对但参数值不对，比如该填数字的填了字符串；三是工具选择错误，该调 A 工具却调了 B 工具。

针对这三类错误，我做了三层防护：

第一层：提示词约束。在系统提示词里明确工具调用的格式要求，并给出正例和反例。这层能解决大部分格式错误。

第二层：输出解析容错。写一个健壮的 JSON 解析器，能处理常见的格式问题，比如尾随逗号、单引号、未转义字符。解析失败时，把错误信息返回给模型让它重新生成。

第三层：参数校验。工具执行前先校验参数类型和范围，不合法就拒绝执行并返回错误提示。这层能防止参数错误导致的意外行为。

def safe_parse_tool_call(raw_output): # 尝试直接解析 try: return json.loads(raw_output) except json.JSONDecodeError: pass # 尝试修复常见问题 fixed = raw_output.strip() fixed = re.sub(r',\s*}', '}', fixed) # 去掉尾随逗号 fixed = re.sub(r',\s*]', ']', fixed) fixed = fixed.replace("'", '"') # 单引号转双引号 try: return json.loads(fixed) except json.JSONDecodeError: return None # 解析失败，触发重试

实测下来，三层防护把工具调用的成功率从 70% 左右提升到了 95% 以上。剩下的 5% 主要是模型对复杂业务逻辑的理解偏差，这个只能靠优化工具描述和 Skills 设计来改善。

3. 实操过程与核心环节实现

3.1 离线环境的依赖打包与部署

内网部署最大的体力活是依赖打包。外网环境下pip install一条命令搞定的事，内网要手动下载所有依赖的 wheel 包，拷贝进去再离线安装。

我的做法是在一台和外网环境一致的机器上，用pip download把所有依赖下载到本地目录：

pip download -r requirements.txt -d ./offline_packages --platform manylinux2014_x86_64 --python-version 310 --only-binary=:all:

这里有几个坑要注意。平台标签要匹配，内网机器的操作系统和 Python 版本必须和外网下载时指定的平台一致，否则 wheel 包装不上。有些包没有预编译 wheel，需要下载源码包并在内网机器上编译，这就要求内网机器有完整的编译工具链。

模型文件的搬运更麻烦。一个 7B 参数的模型，量化后也有 4GB 左右。我用移动硬盘拷贝，拷贝前先做分卷压缩和校验，避免传输过程中损坏。

部署脚本我写成了一个一键安装的 shell 脚本，自动完成依赖安装、模型加载、数据库初始化、服务启动。这个脚本后来成了团队的标准部署工具，新机器上线从半天缩短到半小时。

3.2 MCP Server 的实现与调试

MCP Server 我用 Python 实现，核心是处理标准输入输出的 JSON-RPC 消息。协议本身不复杂，但调试起来比较麻烦，因为 stdio 通信看不到中间过程。

我的调试方法是：在 Server 端加详细的日志，把收到的每个请求和发出的每个响应都写到日志文件。同时写一个测试客户端，可以手动发送请求来验证 Server 的行为。

class MCPServer: def __init__(self): self.tools = {} def register_tool(self, name, description, parameters, handler): self.tools[name] = { "description": description, "parameters": parameters, "handler": handler } def handle_request(self, request): method = request.get("method") if method == "tools/list": return {"tools": [ {"name": n, "description": t["description"], "parameters": t["parameters"]} for n, t in self.tools.items() ]} elif method == "tools/call": tool_name = request["params"]["name"] arguments = request["params"]["arguments"] if tool_name not in self.tools: return {"error": f"Unknown tool: {tool_name}"} try: result = self.tools[tool_name]["handler"](**arguments) return {"content": [{"type": "text", "text": str(result)}]} except Exception as e: return {"error": str(e)}

调试过程中发现一个关键问题：工具执行超时。有些查询可能耗时较长，如果 Server 一直不返回，主进程会以为卡死了。我的解决方案是给每个工具设置超时时间，超时后返回错误信息而不是无限等待。

3.3 流式输出的实现细节

Agent 的响应需要流式输出，否则用户要等很久才能看到结果。流式输出分两个层面：模型生成 token 的流式输出，和工具执行结果的流式推送。

模型层面的流式输出，本地推理服务一般都支持，通过 SSE 或 WebSocket 推送 token。我用的推理服务支持 OpenAI 兼容的流式接口，直接对接就行。

工具执行结果的流式推送要自己实现。当一个 Skill 内部依次调用多个工具时，每完成一步就把中间结果推送给前端，让用户看到进度。这个通过一个事件队列实现：Skill 执行过程中往队列里放事件，主进程从队列取事件并推送给前端。

class StreamingSkill(BaseSkill): def execute(self, **kwargs): self.emit("progress", "开始查询数据...") data = self.call_tool("query_database", sql="...") self.emit("progress", f"查询到 {len(data)} 条记录") self.emit("progress", "正在生成分析...") analysis = self.analyze(data) self.emit("progress", "分析完成") return analysis

实操心得：流式输出的粒度要适中。太细了，前端频繁更新影响性能；太粗了，用户感觉不到进度。我的经验是每个有意义的步骤推送一次，比如"开始查询""查询完成""开始分析""分析完成"。

3.4 十万条数据查询的性能优化实录

十万条数据在 SQLite 里不算多，但如果查询写得不好，照样能跑出几秒的延迟。我做了几轮优化，把典型查询从 800ms 降到了 30ms 以内。

第一轮优化：加索引。这是最立竿见影的。Agent 最常查询的字段组合是"产线+日期"，我建了复合索引：

CREATE INDEX idx_line_date ON quality_data(line_name, record_date);

加索引后，查询从 800ms 降到 50ms。

**第二轮优化：避免 SELECT ***。模型生成的 SQL 经常是SELECT *，把整行数据都拉出来。我通过工具描述约束模型必须指定列名，同时在后端加了一层拦截，检测到SELECT *就自动改写或拒绝。

第三轮优化：查询结果缓存。有些查询是重复的，比如"今天的总产量"可能被问很多次。我在内存里加了一层 LRU 缓存，相同 SQL 在 5 分钟内直接返回缓存结果。

第四轮优化：分页查询。对于可能返回大量结果的查询，强制加 LIMIT 和 OFFSET，分批返回。Agent 需要更多数据时再查下一页。

优化前后的对比：

4. 常见问题与排查技巧实录

4.1 工具调用失败的排查思路

工具调用失败是最常见的问题，表现是 Agent 说"我无法完成这个操作"或者直接报错。排查思路我总结成了一个流程：

第一步：看审计日志。tool_call_log表里记录了每次调用的详细信息。先看有没有调用记录，如果没有，说明模型根本没触发工具调用，问题在提示词或模型理解上。如果有记录但 success=0，看错误信息是什么。

第二步：复现问题。用相同的参数手动调用工具，看是否能成功。如果手动调用成功但 Agent 调用失败，说明是参数传递的问题。如果手动也失败，说明是工具本身的问题。

第三步：检查参数。对比 Agent 生成的参数和工具期望的参数，常见问题是类型不匹配、必填参数缺失、参数值超出范围。

第四步：优化描述。如果确认是模型理解偏差，回头优化工具描述，把容易混淆的地方写清楚。

常见问题速查表：

4.2 模型输出不稳定的应对策略

本地模型的一个通病是输出不稳定，同样的输入可能得到不同的输出。这在工具调用场景下很致命，因为格式错误会导致整个流程失败。

我的应对策略是降低对模型输出稳定性的依赖。具体做法：

结构化输出约束。在提示词里明确要求模型按固定格式输出，并给出模板。比如要求工具调用必须包裹在特定的标记里，方便解析。

重试机制。解析失败时，把错误信息返回给模型，让它重新生成。一般重试 2-3 次就能成功。重试时要调整温度参数，降低随机性。

降级方案。如果重试多次仍失败，降级到规则匹配。比如用户问"产量是多少"，规则匹配直接触发查询产量的工具，不依赖模型判断。

人工兜底。对于关键操作，如果 Agent 无法完成，提供人工介入的入口。运维人员可以手动执行操作，Agent 记录操作过程用于后续学习。

4.3 内网环境的特殊坑与规避

内网环境有一些外网遇不到的坑，我踩过几个印象深刻的。

时间同步问题。内网机器的时间可能和外网不一致，导致模型生成的时间相关查询出错。我的解决方案是在 Agent 启动时强制同步一次时间，并在提示词里注入当前时间。

字符编码问题。内网的一些老系统用 GBK 编码，而 Agent 默认用 UTF-8，数据交换时出现乱码。解决方案是在数据接入层做编码转换，统一转成 UTF-8。

磁盘空间问题。模型文件、日志文件、数据库文件加起来占用不小，内网机器磁盘空间有限。我加了日志轮转和数据库清理策略，定期归档旧数据。

依赖版本冲突。内网无法在线解决依赖冲突，只能手动处理。我的做法是用虚拟环境隔离，每个组件独立环境，避免相互影响。

避坑技巧：内网部署前，先在一台和外网隔离的测试机上完整走一遍部署流程，把所有依赖和配置问题提前暴露出来。直接在生产环境部署，出问题排查起来非常痛苦。

4.4 Skills 测试与质量保障

Skills 的质量直接决定 Agent 的可用性。我建立了一套测试流程，每个 Skill 上线前必须通过。

单元测试：测试 Skill 的每个步骤，确保单独执行时正确。用 mock 数据模拟工具返回，验证 Skill 的逻辑分支。

集成测试：测试 Skill 和真实工具的配合，确保工具调用参数正确、结果处理正确。

端到端测试：模拟用户提问，验证 Agent 能否正确选择 Skill 并完成整个流程。

回归测试：每次修改 Skill 或工具后，跑一遍全部测试用例，确保没有破坏已有功能。

测试用例我维护在一个 YAML 文件里，包含输入、期望的工具调用序列、期望的输出。测试脚本自动执行并对比结果。

- name: 查询本周产量 input: "本周产线A的产量是多少" expected_tools: - name: query_database params_contain: ["产线A", "本周"] expected_output_contains: ["产量", "件"]

这套测试流程帮我发现了不少问题，比如某个 Skill 在数据为空时的处理逻辑有 bug，某个工具的参数校验太严格导致正常查询被拒绝。上线前发现总比上线后被用户发现好。

5. 性能调优与扩展性思考

5.1 响应延迟的优化实践

Agent 的响应延迟由三部分组成：模型推理时间、工具执行时间、网络传输时间。内网环境下网络传输可以忽略，主要优化前两者。

模型推理时间是大头。7B 模型在 24GB 显存的显卡上，生成 100 个 token 大约需要 1-2 秒。优化手段包括：使用量化模型减少显存占用和计算量、开启批处理提高吞吐、缓存常见问题的回答。

工具执行时间通过前面说的索引优化和缓存已经降下来了。还有一个优化点是并行执行。如果一个 Skill 内部有多个独立的工具调用，可以并行执行而不是串行。比如查询本周数据和上周数据是独立的，可以同时查。

import concurrent.futures def execute_parallel(self, tasks): with concurrent.futures.ThreadPoolExecutor(max_workers=4) as executor: futures = [executor.submit(task) for task in tasks] return [f.result() for f in futures]

实测下来，并行执行把某些 Skill 的耗时从 3 秒降到了 1.5 秒。

5.2 多用户并发场景的应对

一开始系统只支持单用户，后来要支持多个运维人员同时使用。并发带来的问题是资源竞争：模型推理排队、数据库锁冲突、内存不足。

模型推理排队：本地推理服务的并发能力有限，我加了一个请求队列，按优先级调度。简单查询优先，复杂分析排队。

数据库锁冲突：SQLite 的写锁是全局的，多个会话同时写日志会冲突。开启 WAL 模式后，读操作不阻塞，写操作串行化。对于日志写入，我用了批量写入策略，攒够一定数量再一次性写入，减少锁竞争。

内存不足：每个会话都维护上下文，内存占用随会话数增长。我加了会话超时机制，闲置超过 30 分钟的会话自动释放内存。

5.3 后续扩展方向

这套系统目前能满足基本需求，但还有不少可以扩展的地方。

多模型支持：目前只接了一个本地模型，后续可以接入多个模型，根据任务类型选择最合适的。简单查询用小模型快速响应，复杂分析用大模型保证质量。

知识库增强：目前 Agent 主要靠 SQL 查询和预定义 Skill，后续可以接入本地文档知识库，支持基于文档的问答。这需要引入向量检索能力。

自动化工作流：目前 Skill 是预定义的，后续可以支持用户自定义工作流，通过可视化界面编排工具调用序列。

移动端适配：目前只有 Web 界面，后续可以适配移动端，让运维人员用手机就能查询和操作。

这套系统从立项到上线用了大约两个月，其中大部分时间花在工具链搭建和调试上。真正核心的 Agent 逻辑代码量并不大，但周边的工程化工作很繁琐。如果你也在做类似的项目，我的建议是：先把工具链跑通，再优化 Agent 效果。工具链不通，Agent 再聪明也干不了活。