隔离内网环境下AI Agent落地实践：从LangGraph编排到vLLM私有化部署

在隔离内网里做 AI Agent，和你在公网环境写 demo 完全是两码事。几天前我刚把一个 Agent 项目从个人开发机搬到客户的隔离内网里，第一天就吃了大亏：模型必须走本地私有化推理，工具调用全部指向内网接口，连装一个 Python 依赖都得先解决离线源的问题。这篇就是完整复盘，从需求拆解、架构设计、模型底座选型，到 LangGraph 编排、并发压测和问题排查，全部基于我实际落地的流程。看完你至少能少走我三分之一的弯路。

这套系统适合谁参考？两类人：一是企业内部正在做"基于私有化大模型 + 自动化工具"的 AI 应用工程师，二是准备把 Agent 从公网迁移到隔离内网的团队负责人。它解决的核心痛点是"模型怎么在断网环境下跑起来、Agent 怎么安全地调用内网服务、并发上去后怎么不雪崩"。

1. 项目背景与需求拆解

1.1 为什么要在隔离内网里做 Agent

很多业务系统天然就不能连公网。金融、运营商、政务、大型制造企业的核心生产区，数据出不去，外部大模型接口自然也用不了。但业务方又想要"AI 能直接干活"——查库存、看工单、读报表、生成值班记录，这些诉求全靠人肉完成太浪费，于是"隔离内网 + AI Agent"就成了唯一出路。

所谓隔离内网，不只是没外网那么简单。它是三层约束层层叠加：第一，网络层断外，所有通信只能在内网完成；第二，模型层必须私有化部署，不能调用任何线上大模型 API；第三，工具层能对接的只有企业内部服务，比如统一认证、数据库、工单系统、监控平台。Agent 的本质是"大模型 + 工具调用"，在隔离内网里，工具从公网 API 换成了内网服务，大模型从 SaaS 换成了本地推理，架构思路要全部重来。

我这次的项目是给一个生产区搭"智能运维助手"：员工用自然语言提问"最近两小时支付接口的报错率是多少"、"帮我把这几个工单标成高优先级"，Agent 负责理解意图、拆解任务、调内网监控 API 和工单 API、最后汇总答案。生产区是典型的隔离内网，对数据不出域有硬性要求，模型推理必须在自己机房完成。

1.2 隔离环境带来的四个硬约束

这四条约束看起来是环境问题，实际上决定了整个技术选型，我把它们列在最前面：

• 模型必须本地化部署。推理依赖 GPU 资源，能跑多大模型、用不用量化、并发上限是多少，全由机房硬件决定，而不是由云厂商决定。
• 依赖必须离线安装。所有 Python 包、模型权重、向量库都要先在外网环境准备好，再想办法传进去，pip install 直接从 PyPI 拉包是行不通的。
• 工具调用只能面向内网服务。不能依赖回调公网 webhook，Agent 的工具集必须全部封装成对内网服务的安全调用。
• 并发能力受限于本地资源。公网可以弹性扩容，隔离内网一般就固定几台 GPU 服务器，必须把吞吐做好，否则用户一多就卡死。

这四个约束里，最容易翻车的是第三条。业务方会觉得"反正内网安全，工具随便调"，但实践经验告诉我，内网工具恰恰是 Agent 事故高发地：LLM 输出不可控，参数校验没做好，一句话就能让 Agent 去调接口把数据批量改坏。后面我会单独讲工具层的安全设计。

2. 整体架构设计与技术选型

2.1 Agent 主流程：从用户问题到任务闭环

Agent 系统的核心不是模型多聪明，而是"怎么把一次问答变成一次可靠的任务执行"。隔离内网环境里没有兜底的公网服务，主流程设计必须保守且可追踪。我采用的是经典的"理解-规划-执行-综合"四段式：

用户问题进来后，先做意图识别和必要的信息抽取，确定是否需要调用工具，还是直接问答就行。如果需要工具，Agent 生成一份简短的执行计划，按顺序或并行调用工具，每步拿到结果后决定下一步动作，最后把多步结果汇总成用户可以理解的回答。所有过程记录到日志里，方便事后审查——这在隔离内网尤其重要，操作要留痕。

流程本身不复杂，难在"可靠"。LLM 在规划步骤时可能给出不存在的工具名，工具返回的数据可能不符合预期格式，内网某个服务可能正好在发布重启。所以我在框架层加了两个机制：一是工具名和参数的严格校验，规划结果必须过一层白名单过滤才能执行；二是对每一步的调用结果做格式校验，失败了就走重试或降级策略，而不是把错误原样抛给用户。

2.2 Agent 框架选型：LangGraph 还是自研编排

架构选型阶段，我们在三条路线之间犹豫了很久。第一条是 LangChain + LangGraph 组合，第二条是 Spring AI，第三条是完全自研编排器。最终选了 LangGraph + FastAPI，原因是它把有状态编排、条件跳转、人工确认这类能力都内置了，不需要我从零造轮子，而且对 Python 技术栈的团队最友好。

我对比过这三条路线的适用场景，整理成一张表供参考：

选 LangGraph 还有一个关键原因：隔离内网环境里调试成本高，重新部署一次要半天，所以框架本身的可靠性和可观测性比炫技重要。LangGraph 自带状态检查点和逐步回放，出问题时我能把 Agent 的完整执行轨迹打出来，这在断网环境里价值巨大。

2.3 模型底座：离线部署的几条可行路线

隔离内网没有公网模型 API，模型底座只能在开源权重 + 本地推理引擎这个组合里选。我的经验是，离线环境首选 Qwen 系列和 GLM 系列，原因有两个：中文能力强、权重在开源社区好找且自带商用许可。代码生成类任务可以再挂一个 CodeLlama 衍生模型。

推理引擎的选择直接影响并发和响应速度。我从轻到重列一下主流路线：

• Ollama + llama.cpp：最轻量，安装方便，适合内网快速验证和单人使用，跑 7B 以下模型够用，但并发和吞吐一般，也不适合大规模提供 OpenAI 兼容服务。
• vLLM + 部分量化模型：工程上最平衡的方案，PagedAttention 和 Continuous Batching 带来的吞吐提升非常明显，官方提供 OpenAI 兼容 API，接 FastAPI 几乎零成本。
• TensorRT-LLM：性能天花板高，但部署复杂度大，需要针对模型专门做引擎构建，隔离内网里换一次模型要折腾好久，适合模型基本固定、追求极致吞吐的团队。

我这次选了 vLLM。原因很直白：拿一台 A100 80G 跑 32B 量化模型就能支撑几十个并发用户，响应速度和稳定性都达标。这里给一个算显存的经验公式：模型参数占用量约等于 参数量 × 每个参数的字节数。FP16 是 2 字节，INT8 是 1 字节，INT4 是 0.5 字节。一个 32B 模型 FP16 需要 64GB 显存，加上激活和 KV Cache，80G 单卡很紧张，所以线上我用 AWQ 量化的 32B 模型，权重降到约 16GB，留出充足 KV Cache 空间，吞吐一下子舒服了。

3. 核心模块实现与实操细节

3.1 模型统一接入层：让上层只认 OpenAI 协议

隔离内网里经常要同时跑多个模型，比如一个 32B 主模型负责复杂对话，一个 7B 小模型负责意图分类和工具规划。如果每个 Agent 都直接连各自的推理服务，切换模型时就改到怀疑人生。所以我在模型层做了一层统一抽象：所有推理服务都暴露 OpenAI 兼容接口，上层代码只面向统一的 client。

vLLM 官方支持 OpenAI 风格 API，启动起来非常方便。服务端起一个模型实例：

vllm serve /data/models/qwen2.5-32b-instruct-awq \ --served-model-name qwen2.5-32b-instruct \ --port 8001 \ --max-model-len 8192 \ --gpu-memory-utilization 0.9

上层 Python 接入就一行指向内网地址的事：

from openai import OpenAI client = OpenAI( base_url="http://192.168.10.20:8001/v1", api_key="internal-not-required", )

再包一层路由，按场景选择模型：

MODEL_ROUTE = { "planner": "http://192.168.10.20:8002/v1", # 7B 小模型，负责规划 "reasoner": "http://192.168.10.20:8001/v1", # 32B 主模型，负责综合 } def get_client(role: str) -> OpenAI: return OpenAI(base_url=MODEL_ROUTE[role], api_key="internal")

实操心得：统一接入层的关键不是代码量，而是约定。团队必须约定所有模型请求都走这一层，任何人不得绕过它直连推理服务。隔离内网里出问题不好热修，规范比技术更管用。另外，vLLM 启动参数里的--max-model-len一定要按实际业务来，设太大 KV Cache 会被撑爆，设太小长文档场景会直接报长度超限。

3.2 工具服务封装与权限管控：Agent 的"手"必须上锁

工具调用是整个 Agent 系统里风险最高的环节。LLM 本质是概率生成，它规划出来的"调库存接口"可能带上了错误的参数，甚至被用户精心构造的提示词诱导去调用错误工具。在隔离内网里，一个内部工单系统、一个配置库，被 Agent 误操作了，影响面比公网还严重，因为内网权限往往更宽。

我采用了两层防护。第一层是工具注册白名单：Agent 能看到的工具枚举完全由代码注册表决定，模型不认识的工具一个都调不到；第二层是参数强校验，每个工具的参数都有 JSON Schema，执行前用 Pydantic 做类型校验，把非预期参数挡在调用链之外。

工具封装我用一个简单的注册表实现：

from pydantic import BaseModel, Field from typing import Dict, Callable TOOL_REGISTRY: Dict[str, dict] = {} def register_tool(name: str, description: str, schema: type[BaseModel]): def decorator(func: Callable): TOOL_REGISTRY[name] = {"description": description, "schema": schema, "func": func} return func return decorator class QueryErrorRateParams(BaseModel): start_time: str = Field(description="开始时间，ISO 格式") end_time: str = Field(description="结束时间，ISO 格式") service: str = Field(description="服务名，必须等于 payment-api") @register_tool("query_error_rate", "查询指定服务在时间范围内的错误率", QueryErrorRateParams) def query_error_rate(params: QueryErrorRateParams): # 内部实现：调监控平台接口 ...

真正执行前必须过一道分发器：

def dispatch(name: str, raw_params: dict): tool = TOOL_REGISTRY.get(name) if not tool: raise ValueError(f"unknown tool: {name}") validated = tool["schema"].model_validate(raw_params) return tool["func"](validated)

两层防护加在一起，效果很直接：即使模型被绕晕了产生不安全的调用想法，到了校验层也会被拦住。注意：工具层永远不要把数据库连接直接暴露给模型。我见过有人图省事写了个execute_sql工具让 Agent 自由查询，结果一个 "查出所有表里包含用户手机号的列名" 的请求直接把生产库扫挂了。工具要按业务动作封装，不要暴露底层操作原语。

3.3 多 Agent 协作与状态编排：从链式到图式

早期版本我用的是简单链式调用：LLM 生成计划 → 逐条执行 → 汇总回答。但真实场景里，用户的需求往往有分支：先查错率，错率高于阈值就自动建工单，低于阈值就只返回报告。这种分支逻辑用链式写开始乱套，我这才切到 LangGraph 的图编排。

LangGraph 的核心是 StateGraph：先定义一个全局状态对象，每个节点读状态、改状态，再用条件边控制走向。我用一个简单例子说明：

from typing_extensions import TypedDict from langgraph.graph import StateGraph, END class AgentState(TypedDict): user_input: str error_rate: float need_ticket: bool final_answer: str def check_error(state: AgentState): # 调用查询工具 state["error_rate"] = dispatch("query_error_rate", { "start_time": "2025-01-01T00:00:00", "end_time": "2025-01-01T02:00:00", "service": "payment-api", }).get("rate") state["need_ticket"] = state["error_rate"] > 0.05 return state def open_ticket(state: AgentState): # 手工建单工具 ... return state def reply(state: AgentState): state["final_answer"] = f"当前错误率 {state['error_rate']:.2%}。" return state graph = StateGraph(AgentState) graph.add_node("check", check_error) graph.add_node("ticket", open_ticket) graph.add_node("reply", reply) graph.add_edge("check", "reply") graph.add_conditional_edges("check", lambda s: "ticket" if s["need_ticket"] else "reply") graph.add_edge("ticket", "reply") graph.add_edge("reply", END)

这套编排方式的最大优势是"状态显式化"。整个任务过程中间变量全部放在 state 里，出问题可以回放、可以断点续跑。隔离内网环境里没有完善的监控大盘，这种可追踪性省了我大量排查时间。

特别提醒：写操作类的工具一定要加人工确认节点。我在 LangGraph 里给"创建工单""修改配置"这类动作加了 interrupt 节点，Agent 执行到这一步会停下来，等人工审核通过才继续。上线后这个机制至少拦下了三四次 Agent 的误操作，是整套系统里性价比最高的一笔投入。

4. 并发、性能与稳定性实战

4.1 并发模型：从直连同步到线程池与任务队列

隔离内网的硬件预算有限，并发设计不能按公网那套"起一百个 Pod 随便造"的思路来。最开始我直接用 FastAPI 的 async 接口去调 OpenAI SDK，结果发现模型推理接口本质是阻塞的——vLLM 内部有自己的调度，但 HTTP 层同步等待时，FastAPI 的事件循环会被卡住，并发一上来接口整体堵死。

后来改成两层方案：快速问答场景，使用 FastAPI + 线程池，把同步的模型调用丢给 ThreadPoolExecutor，每个线程持有一个独立 client，实测能把并发承载从个位数提高到几十；耗时较长的多步 Agent 任务，直接走任务队列，FastAPI 接口只负责接收任务并返回任务 ID，后台 Worker 消费队列执行完整流程，用户用轮询或 SSE 拿结果。

from fastapi import FastAPI from concurrent.futures import ThreadPoolExecutor app = FastAPI() executor = ThreadPoolExecutor(max_workers=16) @app.post("/agent") async def submit_agent_task(payload: dict): loop = asyncio.get_event_loop() task_future = loop.run_in_executor(executor, run_agent_pipeline, payload) # 注意：复杂场景建议直接放任务队列，这里仅为轻量示例

并发上限怎么估算？以一台 A100 80G 跑 32B AWQ 量化模型为例，vLLM 的 decode 阶段总吞吐大约在每秒 1500 到 2500 token 左右，这个数字取决于 KV Cache 和模型结构。假设每次请求平均生成 600 个 token，那理论每秒能处理 3 到 4 个请求，一分钟大概 200 个左右。但这只是理想值，实际还要刨掉 prefill 的计算消耗和业务处理时间。所以给业务线的承诺，我会打五折：单实例支撑 100 个并发以内的 Agent 问答，超出就上多实例 + 负载均衡。

4.2 流式输出与超时控制：体验和稳定性必须一起设计

隔离内网用户对响应速度的耐心比公网用户更差——他们习惯了内部系统"等几秒"的风格，但 Agent 动辄要生成几百字，如果全部等完整再返回，一个 30B 模型生成 800 token 可能要 20 秒，交互完全不可用。流式输出是必选项。

我用 SSE（Server-Sent Events）把模型 token 逐字推给前端，FastAPI 实现很简单：

from fastapi.responses import StreamingResponse @app.post("/chat/stream") async def chat_stream(payload: dict): async def event_generator(): stream = client.chat.completions.create( model="qwen2.5-32b-instruct", messages=payload["messages"], stream=True, ) for chunk in stream: if chunk.choices[0].delta.content: yield f"{chunk.choices[0].delta.content}" return StreamingResponse(event_generator(), media_type="text/plain")

流式输出的同时，超时控制要分级。工具调用类操作我设置 10 秒内必须返回结果；模型首次 token 的等待时间（TTFT）设定 15 秒上限；整体 Agent 流程控制在 120 秒内，超过就返回"任务处理中，请稍后查看"。分级超时能让用户感知到系统"还在跑"，而不是莫名卡死。

4.3 容错与重试：别让一次偶发拖垮整个会话

隔离内网服务虽然网络稳定，但模型推理偶尔会超时，内网工具在发布窗口也会抖动。Agent 流程是多步长链路，任何一步失败都可能让整个任务报废。我总结了三条容错经验：

• 指数退避重试：模型 API 返回 503 或超时，退避 1 秒、2 秒、4 秒重试三次，最多三次，再失败就走降级路径。这个策略在 vLLM 偶发排队时很管用。
• 工具调用必须幂等：这是踩坑踩出来的教训。建工单这个工具，第一次调用超时了，但服务端其实已经建单成功；重试一次，就建了两张重复工单。后来所有写操作工具都加上了幂等键——客户端生成请求 ID，服务端按请求 ID 去重，重试只会返回第一次的结果，不会重复执行。
• 降级路径：大模型挂了，至少让用户能拿到提示和入口。有一次 32B 模型因为显存溢出挂了，我把流量自动切到 7B 模型，虽然回答质量明显下降，但至少任务没中断。

容错设计的原则很简单：隔离内网没有"外部云服务帮我兜着"的选项，每个环节都要想到"如果这里挂了，接下来怎么办"。

5. 常见问题与排查实录

5.1 问题一：Agent 级联超时，根因是 prefill 太长

现象是用户提问后，Agent 在"规划"阶段就卡了 20 多秒，整体流程屡屡超时。一开始我以为是模型算力不够，排查后才发现根因是 prefill 太长：为了让模型准确，我把工具描述、系统提示词、历史上下文全塞进 prompt，一次性进来三千多 token，每个请求都要先跑一遍长文 prefill，推理服务的吞吐被严重拖低。

解决思路是"减负"和"分流"：系统提示词里的工具说明从五千字精简到两千字，只留下工具名、参数、行为边界，详细说明挪到工具层；意图分类这种简单任务单独走 7B 小模型，只需要短 prompt，把主模型的 prefill 压力降下来。优化后 TTFT 从 15 秒降到了 3 秒以内。

5.2 问题二：离线依赖安装反复折腾，干脆锁定版本走全离线

隔离内网环境没法直接 pip 装包，我们最初的做法是人工拷 wheel 包，结果缺一个依赖就多一轮传输。后来我改用"中转机 + 全量 wheelhouse"方案：在能联网的中转机上，用与生产机相同的操作系统和 Python 版本，执行

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

然后把整个 wheelhouse 目录通过内部介质传进隔离内网，在目标机器上执行

pip install --no-index --find-links=./wheelhouse -r requirements.txt

注意：中转机的系统和 Python 版本必须与目标机一致，否则二进制兼容性会坑人。另外 requirements 要用 pip-tools 或 uv 锁定出完整的传递依赖清单，不能只在 requirements 里写顶层包。这一步做好了，后面部署任何 Python 服务都能用到同一套离线源，效率翻倍。

5.3 问题三：LLM 规划出危险工具调用，被白名单拦住了

现象是用户用自然语言引导 Agent"直接修改所有低优先级工单的优先级"，模型规划结果里出现了update_all_tickets_batch这个工具。这个工具其实没注册，但如果不做白名单过滤，LangGraph 可能会尝试复用类似的通用 SQL 工具去执行，那后果就严重了。

我们在分发器里加了双重保险：第一重，工具名必须存在于注册表，否则直接拒绝；第二重，参数值会用 Pydantic 校验，比如update_ticket_priority的 ticket_id 字段只接受长度为 6 位到 10 位的工单编号，任何不符合格式的输入都走异常处理。经验是：永远不要假设 LLM 会遵守系统提示词里的行为规则，校验层必须独立存在。

5.4 常见问题速查表

6. 实测数据与个人感想

6.1 一组有代表性的压测数据

系统上线前我做了一轮内部压测，压测场景是"查询错率 + 生成报告 + 按需建工单"的混合流程，模型为 32B AWQ 量化部署在单张 A100 80G 上。数据如下：

这个数据印证了一个判断：单卡单模型在 50 并发以内是可用的，超过就要考虑拆分实例或对任务做优先级队列。业务侧最终给的并发预估在 30 左右，所以单实例就扛住了。

6.2 我踩过的几个坑和体会

整个项目下来，最有价值的一条体会是：隔离内网里做 Agent，工程纪律比技术炫技重要十倍。公网环境里出问题可以快速打开文档、查社区、换一个库试试；隔离内网里每一步都要先想在前面，版本锁定、工具白名单、幂等重试、人工确认，这些看起来"不酷"的机制，才是系统能长期稳定跑下去的根基。

另外一个体会是，不要把 Agent 想得太智能。它本质还是"LLM 做规划 + 工具做执行"的自动化流水线，模型会在你意想不到的地方出幺蛾子。我在上线后陆续收到过几次用户的反馈，比如"帮我查一下昨天的告警"会被理解成"查昨天全天的告警"而不是"最近 24 小时"，这种语义歧义问题只能靠不断积累的用户反馈去微调提示词和工具描述，没有一劳永逸的解法。

最后再分享一个实用小技巧：给 Agent 的所有外部调用都加一行 request-id 关联日志。用户在群里反馈"刚才那个任务结果不对"，你只要拿到时间点，就可以顺着 request-id 把模型的输入、规划步骤、工具调用参数、每一步耗时完整回放出来。这个习惯帮我省了无数次排查成本，强烈建议所有做 Agent 工程的人都从第一天就做起来。