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 注册表的存储。协议层是 MCPModel 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第一个要解决的问题是传输方式。外网常用的 SSEServer-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( nameweekly_report, description生成指定产线指定周的质检周报, parameters{ line_name: {type: string, description: 产线名称}, week_offset: {type: integer, description: 周偏移0表示本周-1表示上周} } ) class WeeklyReportSkill(BaseSkill): def execute(self, line_name, week_offset0): # 第一步查询本周数据 current_data self.call_tool(query_database, sqlfSELECT ... WHERE line{line_name} AND week{week_offset}) # 第二步查询上周数据做对比 previous_data self.call_tool(query_database, sqlfSELECT ... 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_modeWAL写入性能提升明显而且读操作不会被写操作阻塞。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: fUnknown 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 需要更多数据时再查下一页。优化前后的对比优化措施典型查询耗时说明无优化800ms全表扫描SELECT *加索引50ms复合索引命中指定列名35ms减少数据传输加缓存5ms缓存命中时分页查询30ms单页100条4. 常见问题与排查技巧实录4.1 工具调用失败的排查思路工具调用失败是最常见的问题表现是 Agent 说我无法完成这个操作或者直接报错。排查思路我总结成了一个流程第一步看审计日志。tool_call_log表里记录了每次调用的详细信息。先看有没有调用记录如果没有说明模型根本没触发工具调用问题在提示词或模型理解上。如果有记录但 success0看错误信息是什么。第二步复现问题。用相同的参数手动调用工具看是否能成功。如果手动调用成功但 Agent 调用失败说明是参数传递的问题。如果手动也失败说明是工具本身的问题。第三步检查参数。对比 Agent 生成的参数和工具期望的参数常见问题是类型不匹配、必填参数缺失、参数值超出范围。第四步优化描述。如果确认是模型理解偏差回头优化工具描述把容易混淆的地方写清楚。常见问题速查表问题现象可能原因解决方法模型不调用工具工具描述不清晰优化描述增加使用示例JSON 格式错误模型输出不稳定加解析容错失败重试参数类型错误描述未明确类型在描述中强调类型要求工具执行超时查询数据量过大加 LIMIT优化索引返回结果为空SQL 条件错误检查 WHERE 条件加日志上下文超限历史对话过长压缩历史限制轮次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_workers4) 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 再聪明也干不了活。