1. 为什么“隔离内网 AI Agent”是个真问题而不是伪需求先把场景说清楚。所谓隔离内网指的是那种物理上或逻辑上与公网断开、只能通过跳板机或专线做有限数据交换的网络环境。金融、能源、制造、医疗、政务相关的研发团队大概率都待过这种环境。它的核心特征就三条出不去、装不上、传不进。出不去是指没法直接访问外部模型 API装不上是指 pip、npm、apt 这些包管理器基本处于半瘫状态传不进是指你没法随手把一个几十兆的模型权重或者依赖包拖进去。而 AI Agent 这个东西天然是个“吃生态”的活儿。它要调模型、要跑工具、要读文件、要连数据库、要做多轮编排。你在公网上搭一个 Agent可能半小时就跑起来了因为pip install一句话解决所有依赖。但在隔离内网里这一句话可能要变成三天的手工搬运。所以这个标题的价值不在于“教你在内网跑个模型”而在于把 AI Agent 当成一个正经的工程系统来对待解决它在受限环境下的依赖管理、协议适配、工具调用、并发承载和可观测性问题。热词里出现的 MCP、Skills、harness、并发、中台这些词本质上都指向同一件事Agent 不是一个脚本是一套需要工程化的服务。这篇文章适合三类人看。第一类是在内网环境里做 AI 落地的工程师你需要一套能落地的搬运和部署方法论。第二类是想理解 Agent 工程化的开发者公网环境你也能用这套思路做架构。第三类是技术负责人你要评估在内网做 Agent 到底要投入多少人力。我先把结论摆前面隔离内网做 AI Agent难点从来不是模型本身而是依赖闭环、协议桥接和并发治理这三件事。下面按这个逻辑一层层拆。2. 依赖闭环把公网生态“搬”进内网的正确姿势2.1 先搞清楚你到底缺什么别盲目搬很多人一进内网就慌觉得什么都要搬。实际上你得先做一次依赖盘点。AI Agent 的依赖大致分四层层级典型内容是否必须进内网搬运难度运行时Python/Node 解释器、CUDA 驱动必须低离线包即可框架层LangChain、LangGraph、Spring AI必须中依赖树深模型层本地权重或内网推理服务必须高体积大工具层MCP Server、Skills 插件、浏览器驱动按需中高常含二进制盘点的关键是区分“编译期依赖”和“运行期依赖”。比如browser use这类工具编译期可能只需要一个 Python 包但运行期它要拉起一个 Chromium 内核这个内核才是真正难搬的东西。我见过太多人把 pip 包搬进去了结果一跑就报找不到浏览器可执行文件。提示在内网做依赖盘点时建议用pip download或npm pack在公网机器上把整个依赖树拉下来而不是只拉顶层包。顶层包往往只是冰山一角。2.2 离线依赖仓库的搭建别用 U 盘硬拷最土的办法是 U 盘拷 wheel 包但依赖一多就会陷入“装 A 缺 B装 B 缺 C”的死循环。正确做法是在内网搭一个私有包索引。Python 这边用devpi或者简单的pip index静态目录都行。核心操作是在公网机器上执行pip download -r requirements.txt -d ./packages把整个目录同步进内网然后在内网起一个静态 HTTP 服务指向这个目录配置pip.conf的index-url指向它。# 公网侧拉全量依赖 pip download -r requirements.txt -d ./offline_packages --platform manylinux2014_x86_64 --python-version 311 --only-binary:all: # 内网侧起静态索引 python -m http.server 8080 --directory /opt/offline_packages # 客户端 pip.conf [global] index-url http://内网索引地址:8080/simple trusted-host 内网索引地址这里有个坑--platform和--python-version必须和内网目标机完全一致否则拉下来的 wheel 装不上。我踩过一次公网是 x86 拉包内网是 ARM 服务器全部白干。Node 这边类似用verdaccio搭私有 registry或者更简单把node_modules整个目录打包同步。前端 Skills 相关的依赖很多是纯 JS 包直接拷node_modules反而比走 registry 更省事。2.3 模型权重的搬运与内网推理服务模型层是最重的。如果你用的是开源模型权重动辄几个 G 到几十个 G。搬运策略有两种第一种是分片传输 校验。把权重按 1G 切片用split命令切进内网后用cat合并再用sha256sum校验完整性。这个办法慢但稳适合带宽极差的场景。第二种是内网已有推理服务Agent 只做客户端。这是更推荐的架构。内网通常已经有 GPU 集群跑着推理服务比如 vLLM 或 TGI 部署的模型Agent 只需要通过内网 HTTP 接口调用即可。这样 Agent 本身很轻不需要搬权重。# Agent 侧调用内网推理服务 import requests def call_internal_llm(prompt, modelinternal-qwen): resp requests.post( http://内网推理服务:8000/v1/chat/completions, json{ model: model, messages: [{role: user, content: prompt}], temperature: 0.7 }, timeout60 ) return resp.json()[choices][0][message][content]这个接口要兼容 OpenAI 格式这样你的 Agent 框架LangChain、Spring AI 等几乎不用改代码只改base_url就行。这是内网 Agent 工程里最省事的一个设计决策。3. MCP 与 Skills内网 Agent 的工具调用怎么落地3.1 MCP 到底是什么为什么内网更需要它MCP 是 Model Context Protocol一个让模型和外部工具、数据源通信的协议标准。热词里有人问“MCP 是软件协议还是硬件协议”答案是软件协议而且是应用层的。它解决的核心问题是以前每个 Agent 框架都要自己定义一套工具调用格式换个框架就得重写。MCP 把这个格式标准化了。在内网环境里MCP 的价值反而更大。因为内网的工具数据库、文件系统、内部 API都是私有的你不可能指望公网生态给你现成的适配器。有了 MCP你只需要在内网写一个 MCP Server 暴露这些工具任何支持 MCP 的 Agent 客户端都能直接调用。MCP 的通信方式主要有两种stdio标准输入输出适合本地进程和SSE/HTTP适合跨机器。内网跨机器场景用 HTTP 方式更合适。# 一个极简的内网 MCP Server 示例HTTP 方式 from fastapi import FastAPI from pydantic import BaseModel app FastAPI() class ToolCall(BaseModel): tool: str params: dict app.post(/mcp/tools/call) def call_tool(req: ToolCall): if req.tool query_internal_db: # 实际查询内网数据库 result query_db(req.params[sql]) return {result: result} elif req.tool read_internal_file: with open(req.params[path], r) as f: return {result: f.read()} return {error: unknown tool} app.get(/mcp/tools/list) def list_tools(): return { tools: [ {name: query_internal_db, description: 查询内网数据库}, {name: read_internal_file, description: 读取内网文件} ] }这个 Server 跑在内网Agent 通过内网地址调用它。整个链路不出内网安全可控。3.2 Skills 的工程化管理别把插件写成一次性脚本Skills 这个词在不同语境下含义不同。在 Claude 生态里它指的是一种可复用的能力包在更广义的 Agent 工程里它泛指 Agent 可以调用的工具集。内网环境下Skills 管理最大的问题是版本混乱。我建议在内网建一个 Skills 仓库每个 Skill 是一个独立目录包含三样东西manifest.json描述元信息、handler.py执行逻辑、test.py自测用例。Agent 启动时扫描这个仓库动态加载。// manifest.json { name: internal_report_generator, version: 1.2.0, description: 根据内网数据生成周报, entry: handler.py, params: { date_range: {type: string, required: true}, department: {type: string, required: false} } }这样做的好处是当你要新增一个 Skill只需要往仓库里丢一个目录不用改 Agent 主程序。热词里提到的find skills、skills 推荐、codex skills这些本质上都是在找“现成的能力包”但内网场景下你更需要的是自己造能力包的标准流程。3.3 工具调用的权限与审计内网不是法外之地内网环境容易让人放松警惕觉得“反正出不去随便调”。但 Agent 调用工具是有风险的尤其是涉及数据库写操作、文件删除、内部 API 调用的时候。必须做权限分级。我的做法是给每个 Skill 打上权限标签read、write、admin。Agent 在执行时根据当前会话的权限级别决定能不能调。同时所有工具调用都写审计日志记录谁、什么时候、调了什么、参数是什么、结果是什么。def execute_skill(skill_name, params, session): skill load_skill(skill_name) if skill.permission write and session.level 2: raise PermissionError(当前会话无权执行写操作) log_audit(session.user, skill_name, params) return skill.run(params)这套机制在内网尤其重要因为内网的数据往往比公网更敏感。别等出了事才想起来加审计。4. 并发承载AI Agent 怎么扛住真实流量4.1 先算清楚你的并发瓶颈在哪热词里有人问“AI Agent 怎么扛并发”这个问题得拆开看。Agent 的请求链路通常是接收请求 → 编排决策 → 调用模型 → 调用工具 → 汇总返回。每一环都可能是瓶颈。环节典型耗时瓶颈类型优化手段接收请求毫秒级网络 IO异步框架编排决策毫秒到秒级CPU缓存决策结果调用模型秒到十秒级GPU/网络批处理、流式调用工具毫秒到秒级外部依赖连接池、超时汇总返回毫秒级CPU流式输出绝大多数情况下模型调用是最大瓶颈。一个内网推理服务如果只有一张 GPU并发能力可能就是个位数。这时候 Agent 层做再多优化也没用得从推理服务侧解决。4.2 异步编排别让 Agent 傻等模型Agent 编排层必须用异步。Python 用asyncioJava 用CompletableFuture或 Reactor。核心思路是当一个请求在等模型返回时线程/协程应该去处理别的请求而不是阻塞。import asyncio import aiohttp async def call_llm_async(session, prompt): async with session.post( http://内网推理服务:8000/v1/chat/completions, json{model: internal-qwen, messages: [{role: user, content: prompt}]} ) as resp: return await resp.json() async def handle_request(prompts): async with aiohttp.ClientSession() as session: tasks [call_llm_async(session, p) for p in prompts] return await asyncio.gather(*tasks)这样 10 个请求可以并发发出而不是排队等。但要注意并发数不能无限大否则会把推理服务打挂。需要加一个信号量控制并发上限。sem asyncio.Semaphore(5) # 最多 5 个并发 async def call_with_limit(session, prompt): async with sem: return await call_llm_async(session, prompt)这个上限怎么定取决于推理服务的实际承载能力。我的经验是从 1 开始压测逐步加直到响应时间开始明显上升那个点就是上限。4.3 请求队列与降级扛不住的时候要有兜底真实流量是有峰值的。内网 Agent 服务如果直接暴露给多个业务方调用峰值可能远超日常。这时候需要请求队列 降级策略。队列用 Redis 或者内存队列都行。核心是当并发超过阈值新请求进队列等待而不是直接打挂服务。等待超过一定时间返回降级结果比如“当前繁忙请稍后重试”或者返回缓存结果。from collections import deque import time request_queue deque() MAX_QUEUE_SIZE 100 MAX_WAIT_SECONDS 30 def enqueue_request(req): if len(request_queue) MAX_QUEUE_SIZE: return {error: 队列已满请稍后重试} request_queue.append((time.time(), req)) return {status: queued}降级策略要提前设计好。比如模型调用超时可以降级到规则引擎工具调用失败可以降级到返回“暂时无法获取该信息”。别让一个环节的失败拖垮整个请求。4.4 内网推理服务的横向扩展如果内网有多张 GPU 或者多台推理机器Agent 层需要做负载均衡。最简单的做法是在 Agent 侧维护一个推理服务地址列表轮询调用。import itertools LLM_ENDPOINTS itertools.cycle([ http://gpu-node-1:8000, http://gpu-node-2:8000, http://gpu-node-3:8000 ]) def get_llm_endpoint(): return next(LLM_ENDPOINTS)更成熟的做法是用内网的负载均衡器Nginx、HAProxy统一入口Agent 只认一个地址。这样扩缩容对 Agent 透明。5. 内网 Agent 的可观测性与调试5.1 日志Agent 的日志比普通服务复杂三倍普通服务的日志是线性的请求进来、处理、返回。Agent 的日志是树状的一个请求可能触发多次模型调用、多次工具调用每次调用又有自己的输入输出。如果日志不打全出了问题根本没法查。我的做法是给每个请求分配一个trace_id所有相关的模型调用、工具调用、决策步骤都带上这个trace_id。查询的时候按trace_id聚合就能还原整个执行链路。import uuid import logging def handle_agent_request(user_input): trace_id str(uuid.uuid4()) logger.info(f[{trace_id}] 收到请求: {user_input}) decision make_decision(user_input) logger.info(f[{trace_id}] 决策结果: {decision}) for step in decision.steps: result execute_step(step) logger.info(f[{trace_id}] 执行步骤 {step.name}: {result}) return finalize(decision, trace_id)日志格式建议用 JSON方便后续用 ELK 或者 Loki 做聚合分析。内网环境往往没有成熟的日志平台但至少要把日志落到文件并且按天切割。5.2 链路追踪没有它你就是在盲人摸象如果内网有条件建议上 OpenTelemetry。它能把 Agent 的每个环节做成 span可视化展示耗时分布。这样你一眼就能看出瓶颈在模型调用还是工具调用。from opentelemetry import trace tracer trace.get_tracer(agent) def call_llm(prompt): with tracer.start_as_current_span(llm_call) as span: span.set_attribute(prompt.length, len(prompt)) result do_call(prompt) span.set_attribute(result.length, len(result)) return result没有 OpenTelemetry 的话至少要在代码里手动打点记录每个环节的开始和结束时间。这个投入是值得的因为 Agent 的调试成本远高于普通服务。5.3 内网调试的土办法Mock 与回放内网环境调试 Agent 有个特殊困难你没法随便连外部服务做对比测试。这时候Mock 和回放就很重要。Mock 是指把模型调用和工具调用都替换成固定返回用来测试编排逻辑。回放是指把生产环境的真实请求记录下来在测试环境重放验证修改后的逻辑是否一致。# Mock 模型调用 def mock_llm(prompt): if 生成报告 in prompt: return 这是模拟的报告内容 return 这是模拟的回复 # 回放 def replay(trace_file): with open(trace_file) as f: for line in f: record json.loads(line) result handle_agent_request(record[input]) assert result record[output], f回放不一致: {record[input]}这套机制在内网尤其重要因为内网的变更成本高每次上线都要尽量确保不出问题。6. 从零搭建一个内网 Agent 的最小可行架构6.1 架构选型轻量优先别一上来就上中台热词里提到“AI Agent 中台”我的建议是除非你有多个业务方同时要用否则别一上来就搞中台。中台意味着抽象、意味着通用性、意味着更多的代码和更多的坑。先用最小可行架构跑通一个场景再考虑抽象。最小可行架构包含四个组件Agent 服务接收请求做编排决策调用模型和工具。内网推理服务提供模型能力OpenAI 兼容接口。MCP Server暴露内网工具和数据源。Skills 仓库存放可复用的能力包。这四个组件可以跑在同一台机器上也可以分开部署。初期建议同机部署减少网络复杂度。6.2 技术栈选择Python 还是 Java这是内网团队最常纠结的问题。我的判断标准很简单看团队现有技术栈。如果团队是 Python 背景用 FastAPI LangChain/LangGraph。Python 生态在 AI 领域最全MCP 的官方 SDK 也是 Python 优先。缺点是并发性能一般但内网场景通常并发不高够用。如果团队是 Java 背景用 Spring AI Spring Boot。Java 的并发和工程化能力更强适合做中台。缺点是 AI 生态相对弱一些有些新工具没有 Java 版本。// Spring AI 调用内网模型示例 RestController public class AgentController { private final ChatClient chatClient; public AgentController(ChatClient.Builder builder) { this.chatClient builder .baseUrl(http://内网推理服务:8000) .build(); } PostMapping(/agent/chat) public String chat(RequestBody String message) { return chatClient.prompt() .user(message) .call() .content(); } }我个人的经验是做原型用 Python做产品用 Java。原型阶段快速验证想法Python 效率高产品阶段要长期维护Java 的工程化优势就体现出来了。6.3 部署与运维内网没有 K8s 怎么办很多内网环境没有 K8s甚至没有 Docker。这时候用systemd或者supervisor管理进程就行。关键是做好开机自启、崩溃重启、日志轮转这三件事。# /etc/systemd/system/agent.service [Unit] DescriptionAI Agent Service Afternetwork.target [Service] Typesimple Useragent WorkingDirectory/opt/agent ExecStart/opt/agent/venv/bin/python main.py Restartalways RestartSec5 StandardOutputappend:/var/log/agent/agent.log StandardErrorappend:/var/log/agent/agent.err [Install] WantedBymulti-user.target日志轮转用logrotate别让日志把磁盘写满。内网机器磁盘通常不大日志写满导致服务挂掉是很常见的事故。6.4 一个完整的请求生命周期把上面所有东西串起来一个请求在内网 Agent 里的完整生命周期是这样的业务方通过内网 HTTP 调用 Agent 服务带上用户输入和会话 ID。Agent 服务生成trace_id记录请求日志。Agent 编排层分析输入决定调用哪些工具、是否需要模型推理。如果需要模型推理通过内网 HTTP 调用推理服务带上并发信号量控制。如果需要工具调用通过 MCP 协议调用内网 MCP Server。工具返回结果后可能再次调用模型做汇总。最终结果返回给业务方同时记录完整审计日志。如果任何环节超时或失败触发降级策略返回兜底结果。这个链路里每一步都要有超时控制。内网虽然网络稳定但服务本身可能出问题。没有超时控制的 Agent 就是一个定时炸弹。7. 几个我踩过的坑和对应的解法7.1 依赖版本冲突内网没法pip install --upgrade公网环境遇到版本冲突升级一下就好。内网不行你得重新走一遍搬运流程。所以内网环境要锁定版本用requirements.txt精确到小版本号别用。# 好的做法 langchain0.3.7 langgraph0.2.45 fastapi0.115.0 # 坏的做法 langchain0.3.0同时建议在内网维护一个constraints.txt把所有间接依赖也锁死。这样即使某个包更新了也不会因为间接依赖变化导致环境崩掉。7.2 模型输出格式不稳定内网模型往往比公网模型“笨”内网用的开源模型指令遵循能力通常不如公网大模型。你让它输出 JSON它可能给你输出一段带解释的文字。这时候输出解析要极其宽容。我的做法是写一个健壮的解析器先尝试直接json.loads失败就用正则提取 JSON 片段再失败就调用模型做一次“格式化修复”。import json import re def parse_json_robust(text): try: return json.loads(text) except json.JSONDecodeError: pass match re.search(r\{.*\}, text, re.DOTALL) if match: try: return json.loads(match.group()) except json.JSONDecodeError: pass # 最后兜底返回空结构让上层决定怎么处理 return {error: parse_failed, raw: text}别指望模型每次都输出完美格式工程上要做最坏的打算。7.3 工具调用死循环Agent 自己把自己绕进去了Agent 有个经典问题调用工具失败它重试重试还失败它换个方式重试换方式还失败它又绕回第一种方式。结果就是死循环把资源耗光。解法是加最大步数限制和重复检测。每个请求最多执行 N 步超过就强制终止。同时记录每一步的工具名和参数如果发现连续几步完全一样直接中断。MAX_STEPS 15 history [] for step in range(MAX_STEPS): action decide_next_action(history) signature (action.tool, json.dumps(action.params, sort_keysTrue)) if history.count(signature) 3: return {error: 检测到重复调用已终止} history.append(signature) result execute(action)这个坑我在内网环境踩过两次都是因为模型能力弱决策不稳定导致的。加上限制之后就没再出过问题。7.4 内网时间同步问题日志时间对不上内网机器如果没配 NTP各台机器的时间可能差几分钟甚至几小时。这会导致日志聚合时顺序错乱排查问题极其痛苦。进内网第一件事检查所有机器的时间同步。# 检查时间 date # 配置内网 NTP如果有 timedatectl set-ntp true # 或者手动同步 ntpdate 内网NTP服务器这个看起来是小事但排查跨机器问题时时间不一致会让你怀疑人生。8. 关于内网 Agent 工程化的一点个人体会做了几个内网 Agent 项目之后我最大的体会是内网环境逼着你把工程做扎实。公网环境你可以依赖各种托管服务、依赖自动扩缩容、依赖现成的监控告警。内网什么都没有你只能自己造。这个过程很痛苦但造完之后你对 Agent 系统的理解会比只用公网服务的人深得多。另一个体会是别追求一步到位。我见过团队一上来就想搞一个“支持多租户、支持插件热加载、支持分布式编排”的 Agent 中台结果三个月没跑通一个场景。正确的做法是先跑通一个最小闭环一个模型、一个工具、一个业务场景。跑通之后再逐步加能力。最后说一个具体的技巧在内网维护一个“踩坑文档”。每次遇到问题、解决问题都记下来。内网环境信息闭塞新人进来没有公网那么多博客可查这份文档就是团队最宝贵的资产。我现在的团队这份文档已经积累了上百条记录新项目启动时先翻一遍能避开八成以上的坑。内网 Agent 这个方向技术本身还在快速演进MCP、Skills 这些标准也在不断更新。但底层的工程逻辑是不变的依赖要闭环、协议要标准、并发要可控、问题要可查。把这四件事做好剩下的就是时间问题。