1. 隔离内网跑 AI Agent 的真实处境先把场景说清楚。所谓“隔离内网”指的是开发机和生产机都处在没有公网出口的局域网里能访问的只有内部镜像源、内部 Git 服务和几台跳板机。这种环境在金融、制造、政企项目里非常普遍我最近半年接手的两个项目都是这个形态。标题里的“AI Agent 工程实战”说的就是在这种断网环境下把一套能跑起来的 Agent 系统从零搭出来包括模型接入、工具调用、技能编排、并发扛压这一整套。很多人第一反应是没网怎么用大模型其实内网里通常已经部署了私有化的推理服务比如基于开源权重自己搭的推理集群对外暴露一个兼容 OpenAI 协议的 HTTP 接口。Agent 要做的不是去连外部 API而是把内网的模型服务、内网的工具服务、内网的数据源串起来。这件事的难点从来不是“模型聪不聪明”而是工程层面的协议怎么定、工具怎么注册、并发怎么控、失败怎么兜底。这篇文章适合三类人看。第一类是被派到内网环境做 AI 落地的后端或全栈工程师你手里有模型服务但没有成熟的 Agent 框架第二类是正在评估 MCP、Skills 这类新概念到底能不能落地的技术负责人想知道内网里这套东西跑不跑得通第三类是想自己搭一套 Agent 玩票的开发者内网环境反而逼着你把每个环节都搞明白比直接调云服务学到的东西多得多。我下面讲的所有内容都基于一个前提你有一台能访问内网模型服务的开发机有一个内部 Git有一批需要被 Agent 调用的内部系统。至于模型是哪个厂家的、参数多大不影响工程结构。核心关键词会反复出现AI Agent、MCP、Skills、内网、工程实战这几个词基本就是这套系统的骨架。2. 整体架构设计与选型思路2.1 为什么内网 Agent 不能照搬云端方案云端做 Agent 的典型套路是LangChain 或类似框架 外部大模型 API 一堆 SaaS 工具。这套东西搬到内网会立刻碎掉。外部 API 没了SaaS 工具连不上很多框架默认要联网拉取依赖和模型元数据。我踩过的第一个坑就是 pip 装 LangChain 时它去拉 tiktoken 的编码文件内网直接超时整个初始化卡死。所以内网 Agent 的设计原则第一条是所有外部依赖必须可离线化。模型权重、分词器文件、工具描述、甚至 embedding 模型全部要提前下载好放进内网制品库。第二条是协议要足够简单简单到你用 curl 就能调试因为内网里没有花哨的调试工具出问题只能靠日志和手动请求。第三条是工具注册要解耦Agent 核心不应该硬编码任何具体工具否则每加一个内部系统就要改一次核心代码。基于这三条我最终选的架构是一个轻量的 Agent 调度核心自己写不依赖重型框架 MCP 作为工具接入协议 Skills 作为能力编排层。下面分别说为什么。2.2 MCP 在内网里到底解决了什么问题MCP 是什么通俗讲它是一套“工具说明书”的标准格式。以前你要让 Agent 调用一个内部接口得在代码里写死函数签名、参数说明、返回值解析。MCP 把这套东西标准化成一份描述文件Agent 核心读到这份描述就知道有哪些工具、每个工具要什么参数、返回什么结构。内网里 MCP 的价值特别大因为内部系统太多了。我们那个项目里有工单系统、CMDB、日志平台、发布系统四套东西每套都有自己的接口风格。如果不用 MCPAgent 核心里会堆满各种适配代码。用了 MCP 之后每个系统提供一个 MCP ServerAgent 核心只认协议不认具体系统。加一个新系统就是多注册一个 Server核心代码一行不动。这里有个关键决策MCP Server 用 stdio 模式还是 HTTP 模式。stdio 模式是 Agent 启动子进程通过标准输入输出通信适合本地工具HTTP 模式是 Server 独立部署Agent 通过内网 HTTP 调用适合多实例共享。内网生产环境我强烈建议 HTTP 模式因为 stdio 模式下 Agent 进程崩了工具进程也跟着崩而且没法做独立的健康检查和限流。HTTP 模式虽然多一层网络开销但内网延迟本来就在毫秒级这点开销可以忽略。2.3 Skills 层为什么值得单独抽出来Skills 这个词最近很热但很多人没搞清它和 MCP 的区别。我的理解是MCP 解决“能调用什么”Skills 解决“怎么组合调用”。一个 Skill 是一段可复用的能力编排比如“查工单并总结”这个 Skill内部可能先调工单查询 MCP再调模型做摘要最后调通知 MCP 发出去。这三个 MCP 调用被封装成一个 SkillAgent 只需要说“执行查工单并总结”不用关心底层调了几次工具。内网里抽 Skills 层的好处是可测试性和可复用性。每个 Skill 可以单独写单元测试mock 掉 MCP 调用验证编排逻辑对不对。而且不同业务线可以共享 Skill比如“发通知”这个 Skill 被十几个场景复用。如果不抽这一层所有编排逻辑散落在 Agent 的 prompt 里改一处崩一片。Skills 的存储我建议用文件系统而不是数据库。每个 Skill 一个目录里面放描述文件YAML 或 JSON和可选的实现代码。这样版本管理直接用 Git回滚就是 checkout 上一个 commit内网里没有比 Git 更可靠的版本工具了。2.4 并发模型的选择为什么不用纯异步AI Agent 怎么扛并发这是热词里问得最多的。内网 Agent 的并发压力主要来自两块模型推理的排队和工具调用的等待。模型推理通常是瓶颈因为内网 GPU 资源有限一个请求可能要跑几秒到几十秒。工具调用相对快但内部系统偶尔抽风一个请求卡住十几秒也正常。我一开始用纯 asyncio 异步模型结果发现一个问题模型推理服务本身是同步阻塞的异步请求打过去它也是排队反而把 Agent 进程的事件循环堵住了。后来改成异步 IO 信号量限流 独立线程池跑模型调用的混合模型。具体说Agent 主循环是异步的处理请求接收和工具调用模型调用丢到一个固定大小的线程池里线程池大小等于模型服务的并发能力用一个信号量控制同时在飞的模型请求数超过就排队。这个模型实测下来很稳。线程池大小怎么定我的经验值是模型服务 QPS 乘以平均推理耗时。比如模型服务能扛 4 路并发平均推理 5 秒那线程池设 4 到 6 就够了设太大只会让请求在模型侧排队Agent 侧反而积压内存。信号量初始值设成线程池大小保证不会有过量请求涌向模型。3. 核心细节解析与实操要点3.1 内网模型接入的兼容层怎么写内网模型服务大概率不是标准 OpenAI 接口可能是自研的 HTTP 服务参数名和返回结构都不一样。这时候不要在每个调用点写适配而是写一个兼容层对外暴露统一的 chat 接口对内做协议转换。兼容层要处理几个细节。第一是超时内网模型服务如果卡住Agent 不能无限等我一般设 60 秒超时超时后返回一个明确的错误而不是空字符串。第二是重试模型服务偶发 500 可以重试一次但重试要加退避不能立刻重试把服务打垮。第三是流式输出如果模型服务支持 SSE兼容层要把它转成统一的流式接口因为 Agent 前端通常需要逐字显示。import httpx import asyncio from typing import AsyncIterator class ModelClient: def __init__(self, base_url: str, timeout: float 60.0, max_retry: int 1): self.base_url base_url self.timeout timeout self.max_retry max_retry self._client httpx.AsyncClient(timeouttimeout) async def chat(self, messages: list, stream: bool False): payload self._build_payload(messages, stream) last_err None for attempt in range(self.max_retry 1): try: if stream: return self._stream_chat(payload) resp await self._client.post( f{self.base_url}/v1/chat, jsonpayload ) resp.raise_for_status() return self._parse_response(resp.json()) except Exception as e: last_err e if attempt self.max_retry: await asyncio.sleep(1.5 ** attempt) raise RuntimeError(fmodel call failed: {last_err}) def _build_payload(self, messages, stream): # 这里做协议转换把统一格式转成内网模型要的格式 return {messages: messages, stream: stream}这段代码的关键点是_build_payload和_parse_response两个方法它们是你唯一需要根据内网模型改的地方。其他逻辑通用。我建议把这两个方法单独放一个文件方便不同模型服务切换。3.2 MCP Server 的注册与发现机制内网里 MCP Server 怎么让 Agent 知道有两种做法。一种是静态配置Agent 启动时读一个配置文件里面列出所有 Server 的地址。另一种是动态发现Agent 启动时去一个注册中心拉取 Server 列表。内网环境我推荐静态配置 健康检查的组合。静态配置简单可靠配置文件放 Git 里改配置走正常的发布流程。健康检查是 Agent 定期 ping 每个 Server 的/health接口不健康的 Server 暂时从可用列表里摘掉恢复后自动加回来。这样既避免了注册中心的复杂度又能应对 Server 临时挂掉的情况。配置文件格式我一般用 YAML长这样mcp_servers: - name: ticket url: http://ticket-mcp.internal:8080 timeout: 10 health_path: /health enabled: true - name: cmdb url: http://cmdb-mcp.internal:8081 timeout: 15 health_path: /health enabled: true健康检查的间隔别设太短30 秒一次足够。太短会给内网服务增加无谓压力太长则故障发现不及时。连续三次失败才标记为不健康避免网络抖动误判。3.3 Skills 的描述文件设计一个 Skill 的描述文件要包含哪些字段我的实践是最少这几个名称、描述、输入参数 schema、执行步骤、超时、重试策略。执行步骤是核心每一步声明调用哪个 MCP 工具、参数怎么从上下文取、结果怎么存。name: query_and_summarize_ticket description: 查询工单并生成摘要 inputs: ticket_id: type: string required: true steps: - id: fetch tool: ticket.get_ticket params: id: {{ inputs.ticket_id }} save_as: ticket_data - id: summarize tool: model.chat params: prompt: 请总结以下工单内容{{ steps.fetch.output }} save_as: summary - id: notify tool: notify.send params: channel: ticket-{{ inputs.ticket_id }} content: {{ steps.summarize.output }} timeout: 120 retry: 1这里有个设计要点步骤之间的数据传递用模板变量而不是让 Skill 实现代码去手动取值。模板变量让 Skill 描述文件本身就能表达完整逻辑不需要额外写代码。只有特别复杂的 Skill 才需要写实现代码大部分场景描述文件就够了。3.4 参数校验与错误兜底Agent 调用工具时参数经常是模型生成的格式不一定对。比如模型可能把数字生成成字符串把数组生成成逗号分隔的字符串。所以每个 MCP 工具入口都要做参数校验和类型转换不能直接信任模型输出。我的做法是在 MCP Server 里加一层校验中间件用 JSON Schema 校验参数校验失败返回明确的错误信息让 Agent 有机会重新生成参数。错误信息要具体比如“参数 id 应该是字符串实际收到数字”而不是笼统的“参数错误”。模型看到具体错误后重试成功率会高很多。错误兜底分三层。第一层是工具内部重试适合幂等操作。第二层是 Skill 层重试适合整个编排重跑。第三层是 Agent 层兜底如果 Skill 最终失败Agent 要能给出一个合理的失败回复而不是抛异常给用户。这三层都要有缺一层都会导致用户体验断裂。4. 实操过程与核心环节实现4.1 从零搭建 Agent 核心的完整步骤第一步建项目骨架。目录结构我建议这样分core/放 Agent 调度逻辑mcp/放 MCP 客户端和 Server 管理skills/放 Skill 描述文件和实现model/放模型兼容层config/放配置文件。这个结构清晰新人接手能快速定位。第二步实现 MCP 客户端。核心是三个方法list_tools拉取工具列表call_tool调用工具health_check健康检查。list_tools的结果要缓存不要每次调用都拉一遍缓存过期时间设 5 分钟。工具列表变化不频繁缓存能显著降低内网请求量。class MCPClient: def __init__(self, config): self.servers {s[name]: s for s in config[mcp_servers]} self._tool_cache {} self._cache_ttl 300 async def list_tools(self, server_name: str): now time.time() cached self._tool_cache.get(server_name) if cached and now - cached[ts] self._cache_ttl: return cached[tools] server self.servers[server_name] resp await self._client.get(f{server[url]}/tools) tools resp.json()[tools] self._tool_cache[server_name] {tools: tools, ts: now} return tools async def call_tool(self, server_name, tool_name, params): server self.servers[server_name] resp await self._client.post( f{server[url]}/call, json{tool: tool_name, params: params}, timeoutserver[timeout], ) return resp.json()第三步实现 Skill 执行器。执行器读 Skill 描述文件按步骤顺序执行处理模板变量替换和错误重试。模板替换我推荐用简单的字符串替换而不是完整的模板引擎因为内网环境少一个依赖少一个坑。第四步实现 Agent 主循环。主循环接收用户输入判断该走哪个 Skill执行 Skill返回结果。判断走哪个 Skill 可以用模型做意图识别也可以用规则匹配。内网场景我建议规则优先模型兜底因为规则快且可控模型只在规则匹配不上时才用。4.2 并发压测与参数调优实录搭好之后一定要压测。我用 locust 在内网起 50 个并发用户持续打 10 分钟观察几个指标Agent 进程的 CPU 和内存、模型服务的排队长度、工具调用的 P99 延迟。第一轮压测就发现问题Agent 内存持续上涨10 分钟涨了 2G。排查发现是 MCP 客户端的 HTTP 连接没复用每个请求新建连接连接对象没及时释放。改成全局复用一个 httpx.AsyncClient 后内存稳定在 300M 左右。第二轮压测发现模型服务排队严重P99 延迟到了 30 秒。原因是信号量设太大了设了 20但模型服务只能扛 4 路。改成 4 之后P99 降到 8 秒虽然还是有排队但至少不会雪崩。第三轮压测发现工具调用偶发超时。查日志发现是某个内部系统在高峰期响应慢10 秒超时不够。把那个 Server 的超时单独调到 30 秒其他保持 10 秒。这里的原则是超时要按工具特性分别设置不能一刀切。调优后的参数配置参数初始值调优后说明模型线程池大小204匹配模型服务并发能力模型信号量204与线程池一致工具默认超时10s10s通用工具慢工具超时10s30s内部系统类工具健康检查间隔10s30s降低内网压力工具列表缓存无300s减少重复拉取4.3 日志与可观测性怎么落地内网没有云厂商的 APM可观测性全靠自己。我的做法是结构化日志 本地指标聚合。每条日志是 JSON 格式包含 trace_id、skill_name、step_id、tool_name、耗时、状态。trace_id 贯穿一次完整请求方便串联。指标聚合用一个简单的内存计数器定期 dump 到本地文件。关键指标包括请求总数、成功数、失败数、各 Skill 的 P50/P99 耗时、各 MCP Server 的调用次数和失败率。这些指标不需要 Prometheus 那么重的方案一个定时任务每 60 秒统计一次写文件就够了。日志级别要控制好。DEBUG 级别只在排查问题时开平时用 INFO。工具调用的入参和出参在 INFO 级别只记摘要不记全文避免日志爆炸。全文只在 DEBUG 级别记且要加开关。4.4 灰度发布与回滚机制内网发布没有 Kubernetes 那么方便但灰度还是要做。我的做法是双实例 流量切换。新版本部署到新实例先切 10% 流量过去观察 30 分钟指标正常再逐步加到 100%。流量切换通过内网负载均衡的权重配置实现。回滚要能在 1 分钟内完成。所以每次发布前旧版本的实例不销毁只是权重降到 0。出问题直接把权重切回去秒级回滚。这个策略在内网里特别重要因为内网排查问题慢快速回滚能争取大量时间。Skill 描述文件的变更也要走灰度。因为 Skill 是数据不是代码改错了影响面可能更大。我的做法是 Skill 文件也做版本管理新版本先在一个测试 Agent 上验证验证通过再同步到生产。5. 常见问题与排查技巧实录5.1 模型输出格式不稳定怎么办这是最高频的问题。模型有时候返回 JSON有时候返回带 markdown 代码块的 JSON有时候干脆返回一段自然语言。Agent 解析失败就整个流程断了。我的解法是三层解析。第一层直接 JSON 解析成功就用。第二层用正则提取代码块里的 JSON再解析。第三层用模型自己修复把原始输出丢回模型让它输出标准 JSON。三层都失败才报错。实测下来第一层成功率 70%第二层补到 90%第三层补到 98%。还有一个技巧是在 prompt 里明确要求“只输出 JSON不要任何其他文字”并且给一个示例。这个简单的约束能把第一层成功率提到 85% 以上。5.2 MCP Server 连不上怎么快速定位排查顺序我总结成一张表现象可能原因排查方法连接超时网络不通或服务没起telnet 端口看进程401/403认证配置错检查 token 和 header404路径写错对比 Server 文档500Server 内部错误看 Server 日志返回空参数不对手动 curl 复现内网排查有个技巧先用 curl 手动调一遍确认 Server 本身没问题再查 Agent 侧的配置。很多时候问题出在 Agent 的配置文件里 URL 写错了或者超时设太短。5.3 并发下 Skill 执行串数据这个问题很隐蔽。多个请求同时执行同一个 Skill如果 Skill 执行器用了全局变量存中间结果就会串数据。我踩过一次A 用户的工单摘要发到了 B 用户的频道里非常尴尬。根因是 Skill 执行器的上下文没有隔离。解法是每次执行创建独立的上下文对象所有中间结果存在这个对象里不碰任何全局状态。上下文对象随请求创建、随请求销毁。这个原则要写进代码规范Code Review 时重点检查。5.4 内网时间不同步导致的问题内网机器时间不同步是个容易被忽略的坑。Agent 机器和 MCP Server 机器时间差几秒会导致日志时间戳对不上排查问题时误导判断。更严重的是如果 Skill 里有基于时间的逻辑比如“只处理最近 5 分钟的工单”时间差会导致漏处理或重复处理。解法是所有机器统一走内网 NTP。如果内网没有 NTP至少保证 Agent 和关键 Server 的时间源一致。日志里记录的时间戳统一用 UTC展示时再转本地时区避免时区混乱。5.5 内存泄漏的常见来源内网 Agent 长期运行内存泄漏是慢性杀手。常见来源有三个HTTP 连接没复用、缓存没设上限、事件监听器没注销。HTTP 连接问题前面说过全局复用一个 client 就行。缓存没上限的问题工具列表缓存、模型响应缓存都要设最大条数和过期时间我一般设 1000 条、5 分钟过期。事件监听器的问题如果用了发布订阅模式每次请求注册的监听器要在请求结束时注销否则越积越多。监控内存用简单的定时打印每 5 分钟打一次 RSS 和对象数量。发现持续上涨就 dump 内存快照分析。内网没有专业工具用 Python 自带的tracemalloc就够定位大部分问题。6. 内网 Agent 工程的几条硬经验先说工具选型。内网里能用标准库就用标准库能少一个依赖就少一个依赖。我见过太多项目因为一个不起眼的依赖在内网装不上卡了一整天。httpx 这种纯 Python 的库可以放心用涉及 C 扩展的库要提前在内网验证编译。再说配置管理。所有配置必须外置不能硬编码在代码里。内网环境变化频繁IP 会变、端口会变、认证会变配置外置才能不改代码就适配。配置文件的格式我推荐 YAML可读性好支持注释比 JSON 友好。关于测试内网 Agent 的测试要分三层。单元测试测 Skill 编排逻辑mock 掉 MCP 调用。集成测试测 MCP 客户端和真实 Server 的交互在内网测试环境跑。端到端测试测完整链路用真实模型和真实工具但要用测试数据不能碰生产数据。最后说文档。内网项目人员流动大文档是刚需。每个 MCP Server 要有接口文档每个 Skill 要有说明文档Agent 核心要有架构文档。文档放内网 Wiki 或者 Git 仓库的 docs 目录跟代码一起版本管理。我个人的习惯是代码写完先写文档写文档的过程能发现设计上的漏洞。这套东西我前后迭代了三个版本从最初的一把梭到现在的分层架构最大的体会是内网环境逼着你把每个环节都想清楚没有云服务帮你兜底反而让工程能力真正长出来了。MCP 和 Skills 这些新概念在内网里不是花架子它们解决的是真实的解耦和复用问题。如果你也在内网做 Agent欢迎交流踩坑经验尤其是并发和可观测性这两块永远有优化空间。