☰
隔离内网AI Agent实战:私有化模型+LangGraph+FastAPI部署指南
2026/10/2 12:48:56 网站建设 项目流程

1. 甲方说"不能连外网",AI Agent 还怎么做?

先交代一下背景。半年前我接到一个内部项目,需求很直白:在隔离内网里搭一套 AI Agent 服务,让业务人员用自然语言去查询和操作内部系统中的数据。当时的第一反应是——这不就是接个大模型API,包一个Agent壳子吗?结果等甲方把网络拓扑图甩过来,我才意识到事情没那么简单。整个生产环境是物理隔离的内网,不能访问任何外部API,连用pip装个包都要走审批流程。这意味着,平时熟悉的OpenAI SDK、在线Embedding、云上向量库,全部用不了。

那段时间我查了不少资料,也踩了不少坑,后来逐步沉淀出一套在隔离内网里从零搭建 AI Agent 工程的完整方案。这篇文章不是讲理论,就是一次真实的工程复盘:包括技术选型逻辑、私有化模型部署、Agent工作流编排、服务封装与并发处理,以及在内网环境下操作的具体避坑细节。如果你也遇到类似场景——无论是企业内部系统、涉密项目,还是机房环境——希望这篇能让你少走几步弯路。

先说结论:在隔离内网里做 AI Agent,核心不是模型本身,而是三件事。第一,把模型、依赖、镜像全部以离线方式搬运进去;第二,设计一个能支撑多步推理和工具调用的工作流编排层;第三,用一个能扛住真实并发压力的服务框架把能力暴露出来。下面我按这个顺序展开。

2. 技术栈选型:为什么是私有化模型 + LangGraph + FastAPI

选型这件事,我在内网环境下重新思考了很久。说实话,在外网做 Agent,技术栈选择非常多,比如可以直接用云端 LangChain 生态、或者各种商业 Agent 平台。但在隔离内网里,每个组件都要能够离线运行,而且组件之间的交互不能依赖外部网络,这个约束直接把可选项压缩了一大截。

最终我定的组合是:Qwen 系列开源模型 + LangGraph + FastAPI,外加一个内网向量库和关系型数据库做记忆存储。逐个说一下理由。

模型层,我选的是 Qwen2.5 系列。原因很实际:首先是开源协议比较友好,可以商用部署;其次是中文场景的表现,在同等参数规模的模型里确实能打;更重要的是,它有一整套配套的部署工具,模型权重可以一次性离线导入,不需要任何联网验证。参数量上,我根据实际情况做了区分——复杂任务用 14B 的模型做推理,轻量分类和意图识别用 7B 的模型,后面会细讲。

编排层,为什么是 LangGraph 而不是直接用 LangChain 的 Agent 或者自己硬写状态机?LangGraph 本质上是一个有状态的工作流框架,它把 Agent 的每一次推理、工具调用、结果回填都建模成图上的节点和边。对于隔离内网里的业务场景,我们需要的是可控、可追踪、可恢复的多步流程,而不是一个黑盒式的循环调用。LangGraph 的 StateGraph 模型恰好匹配这个需求,而且它支持 Checkpoint 持久化,这个特性在长耗时任务里非常重要。

服务层,FastAPI 在 Python 生态里几乎是异步服务的事实标准。Python 的 GIL 决定了多线程模型在 CPU 密集场景下很吃亏,但 Agent 服务的核心负载其实是外部IO——等待模型推理、等待工具调用返回。这种场景下,异步IO的优势非常明显。FastAPI 基于 asyncio,加上它原生支持 OpenAPI 文档生成,对内部系统对接非常友好。

有一个很容易被忽略的点,是内网环境下的模型访问端口问题。当时我在 TensorRT-LLM 和 vLLM 之间纠结了一阵,两者都是很成熟的推理框架,最后选了 vLLM,原因是它在离线部署时更方便做显存控制和并发策略,且配置项文档更丰富。大家在实际部署时,一定要确认好推理服务的端口能被应用服务器访问,别只在本地 curl 通了就当成功。

下图是整个系统的架构分层,虽然不是标准的架构图,但可以直观看到各层的位置:

  • 模型层:vLLM 托管的 Qwen2.5-14B 和 7B 两套推理服务
  • 编排层:LangGraph 构建的 Agent 状态机,负责意图识别、任务拆解、工具调度
  • 工具层:内部 API 封装、数据库查询、文档检索三大类工具
  • 服务层:FastAPI 应用,接收 HTTP 请求并转发给 Agent 编排层
  • 存储层:PostgreSQL 存会话与状态,向量库存 Embedding

这个分层的好处是,隔离内网环境下的故障边界非常清晰:模型卡住了查模型层,流程卡住了查编排层,接口超时了查服务层和工具层。线上问题排查效率比大一统的应用架构高很多。

3. 内网离线部署模型:绕过"没网就用不了"的关键操作

模型部署是整个工程里最基础也最坑的一步。在外网环境,pip install transformers然后from_pretrained就能把模型从 HuggingFace 拉下来。但在隔离内网,这条路完全走不通。你必须预先在外网环境把模型权重、依赖库、配置文件全部下载好,再通过移动介质或者审批后的网关传输进去。

3.1 离线依赖的完整搬运清单

我整理了一份离线搬运的检查清单,按这个顺序操作基本能一次通过:

  • 模型权重:通过modelscope或者 HuggingFace 的snapshot_download提前下载完整模型目录,注意要下载全部文件,包括 tokenizer.json、config.json 这些看似不起眼的小文件,少一个都可能启动失败。
  • Python 依赖:在有网环境用pip download把 whl 包全部拉下来,要注意目标机器的 Python 版本和系统架构,比如 Python 3.10 的包不能装到 3.8 上。
  • 容器镜像:内网如果需要用 Docker,提前在能联网的机器上docker pull好 vLLM 和相关组件的镜像,然后通过docker save导出成 tar 包。内网机器上用docker load导入,这个方式比手动配容器快得多。
  • 基础配置:时区配置、代理排除配置、日志输出级别等一些环境变量,提前写到环境变量文件里,避免内网启动时因缺少配置导致默认路径挂掉。

3.2 vLLM 启动参数里的实战要点

vLLM 启动时有一堆参数,但真正需要重点关注的其实就那几个。我用了比较典型的一组参数,可以在隔离内网环境跑得很稳定:

python -m vllm.entrypoints.openai.api_server \ --model /data/models/qwen2.5-14b-instruct \ --served-model-name qwen2.5-14b \ --tensor-parallel-size 4 \ --max-model-len 32768 \ --max-num-seqs 32 \ --gpu-memory-utilization 0.92 \ --port 8001 &

几个容易踩坑的参数,我展开说一下:

tensor-parallel-size是张量并行度,取决于你有几张 GPU。这个参数设大了会占满所有通信带宽,设小了显存放不下模型。我们的服务器是 4 张 A100 80G,所以设成 4,负载均衡和显存占用都比较理想。如果是两张卡,设成 2 就行。

gpu-memory-utilization控制显存利用率。我见过很多人不敢把它调高,默认 0.9 其实还好,如果你同时还要在同一批 GPU 上跑别的推理服务,记得适当降一降,比如 0.8。

max-model-len决定最大上下文长度。内网场景下,业务文档可能很长,上下文窗口不够的话,后面的内容会被截断。但把它调大也要付出代价——KV Cache 占用显存会显著增加。实测定下来,32K 是我们的一个平衡点,既有足够的上下文承载能力,又不会因为窗口过大导致并发度下降。如果你的场景是短文本问答,16K 其实完全够。

3.3 Embedding 模型的部署策略

Agent 工具里的文档检索能力离不开 Embedding。这里我选择部署一个独立的 Embedding 服务,而不是复用生成模型来做向量化。原因很简单:Embedding 的调用频次非常高,每次工具调用可能都要做一次向量化,如果和生成模型混在一起,推理队列会互相阻塞,影响响应速度。

Embedding 模型我用的是 BGE-M3,参数不大,部署在单独的容器里。运行方式和 vLLM 类似,只是模型路径和端口不同。这里要特别注意向量维度的一致性:训练检索索引时用的 Embedding 服务,和线上查询时用的一定要是同一个模型、同一个端口,一旦中间换过模型,新旧向量维度不一致,检索效果会断崖式下降。

3.4 一个验证部署成功的通用方法

模型部署完,怎么确认它是真正可用的?我习惯先做一轮简单的推理验证,再用真实业务请求做测试。验证命令如下:

curl -X POST http://<内网IP>:8001/v1/chat/completions \ -H "Content-Type: application/json" \ -d '{ "model": "qwen2.5-14b", "messages": [{"role": "user", "content": "用一句话介绍人工智能。"}], "max_tokens": 200, "temperature": 0.7 }'

如果返回了内容且没有明显错误日志,说明推理链路通了。然后我会用几个业务上的问题测一下,比如让 Agent 去查这个月某个部门的支出明细。这一步能提前暴露很多模型能力边界问题,比如模型对工具描述的语义理解是否准确,或者对复杂指令的遵循能力够不够。

4. 用 LangGraph 编排 Agent 工作流:从一次对话到完整任务

模型部署只是地基,真正让 Agent"下地干活"的是工作流编排层。这块我花的时间最多,因为隔离内网场景下的任务往往不是简单的单轮问答,而是需要多步骤推理,每一步都可能调用不同的工具,还可能有依赖关系。

4.1 不写状态机,让框架替你管状态

如果在 LangGraph 里建模一个带工具调用的 Agent,我最常用的套路是三步:定义状态、定义节点、定义边。整个过程像画一张有向图,节点是操作,边是流转条件。核心代码如下:

from langgraph.graph import StateGraph, END from typing import TypedDict, List class AgentState(TypedDict): messages: List[dict] current_task: str tool_results: dict def llm_node(state: AgentState): # 调用 vLLM 的 API,生成下一步动作 response = call_llm(state["messages"]) return {"messages": state["messages"] + [response]} def tool_node(state: AgentState): # 解析模型输出中的 tool_call,执行对应工具 result = execute_tool(state["messages"][-1]) return {"tool_results": result, "messages": state["messages"] + [result]} graph = StateGraph(AgentState) graph.add_node("llm", llm_node) graph.add_node("tool", tool_node) graph.add_edge("llm", "tool", condition=lambda s: has_tool_call(s)) graph.add_edge("llm", END, condition=lambda s: not has_tool_call(s)) graph.add_edge("tool", "llm") app = graph.compile()

这段代码看着简单,但里面有两个容易被忽略的设计细节。第一,state["messages"]是整个会话的完整消息列表,每一步节点执行后都要把新产生的消息追加进去,这样才能让模型在下一步推理时看到之前的工具调用结果。第二,has_tool_call这个条件判断是动态路由的关键,模型输出的 tool_call 字段是否存在,决定了流程是继续调工具还是结束输出。

4.2 工具即函数:注册一个能被模型正确调用的接口

LangGraph 的 tool 节点内部,我维护了一个工具注册表,每个工具本质上是一个带 JSON Schema 描述的函数。下面是我封装的一个获取系统运行状态工具的例子:

from pydantic import BaseModel from langchain_core.tools import tool class SystemStatusInput(BaseModel): system_id: str = Field(description="系统唯一标识,如 SYS-1001") @tool def get_system_status(system_id: str) -> dict: """获取指定系统的实时运行状态、资源使用率和告警信息""" result = internal_api.query_status(system_id) return result

这个工具看着朴素,但关键全在描述里。大模型没有读代码的能力,它只能通过函数名、输入参数描述和 docstring 来判断该不该调用这个工具、传什么参数。所以我在每一个工具的 docstring 里都写得特别具体,比如"获取指定系统的实时运行状态、资源使用率和告警信息",而不是"查询状态"这种模糊描述。

原因是:我早期版本里有一个工具描述写得含糊,模型经常在多个相似工具之间乱选。后来我把描述改成"获取容器资源使用率,参数传入容器名称,返回 CPU、内存和磁盘指标",误调用的频率明显下降。这也是一个经验:工具描述里必须包含"什么场景下用它、参数是什么含义、返回什么数据"三要素。

4.3 人工审核节点:隔离内网里的"安全阀"

隔离内网里,Agent 往往要操作的是核心业务数据。模型再强,也有犯错的概率。所以我在关键链路加了一个human_confirm节点,凡是涉及写操作或者影响面比较大的查询,都必须经过人工审核。这个节点在 LangGraph 里实现为中断机制,也就是说当执行流走到这个节点时,Agent 会暂停,等待人在 web 界面上点击"允许"或"拒绝",然后再继续往下走。

有人可能会问,这样会不会拖慢效率?实测发现,大部分查询类操作不需要审核,只有变更类操作才需要,所以真实影响不大。但安全性提升了一个量级,这在企业内部落地时是加分项。

4.4 Checkpoint 持久化:长任务失败不用从头跑

LangGraph 的 Checkpoint 机制是我最欣赏的特性之一。在这个 Agent 里,我每处理一次请求就自动保存一次快照。这意味着,一个任务如果在第 5 步工具调用时挂了,恢复到快照后可以从第 5 步开始继续跑,而不是重新发一遍最初的请求。隔离内网环境下网络不稳定,工具调用容易超时,这个特性让我们的重试成本大大降低。

要注意的是,Checkpoint 需要配置一个存储后端。我用的是 PostgreSQL,因为内网里它最容易部署和维护,而且能承载会话状态。配置方式是在 LangGraph 的compile时传入一个checkpointer,代码大致如下:

from langgraph.checkpoint.postgres import PostgresSaver checkpointer = PostgresSaver.from_conn_string("postgresql://agent_user:pass@pg-host:5432/agent_db") app = graph.compile(checkpointer=checkpointer)

这段配置里有个容易踩的坑:PostgreSQL 的表结构需要提前初始化。如果checkpointer.setup()没执行,运行时会直接报错。我第一次就是漏了这一步,花了不少时间排查。

5. FastAPI 服务封装与"扛并发"的真实体验

编排层搞定之后,Agent 还只是藏在 Python 环境里的一个对象,对外要暴露成可以调用的 HTTP 服务。这一步我用 FastAPI 做封装,并把服务做成一个标准化的生产级应用:包含鉴权、限流、结构化响应、异步处理四个部分。

5.1 一个最小可用的 Agent 服务接口

先看核心接口代码:

from fastapi import FastAPI, Depends, HTTPException from pydantic import BaseModel, Field app = FastAPI(title="Agent API Service", version="1.0.0") class AgentRequest(BaseModel): session_id: str = Field(description="会话标识") user_input: str = Field(description="用户输入") need_human_confirm: bool = Field(False, description="是否开启人工审核节点") class AgentResponse(BaseModel): session_id: str output: str trace_id: str used_tools: list[str] @app.post("/api/v1/agent/run", response_model=AgentResponse) async def run_agent(req: AgentRequest, user: str = Depends(verify_token)): trace_id = generate_trace_id() result = await run_agent_async( session_id=req.session_id, user_input=req.user_input, need_human_confirm=req.need_human_confirm, trace_id=trace_id, user=user ) return AgentResponse(**result)

这里的run_agent_async是异步执行 Agent 任务的函数,内部把 LangGraph 的app.ainvoke包装成一个可等待的任务。注意,我在这里没有直接用run_agent函数名作为路由名,而是用了/api/v1/agent/run,这是为了方便多版本共存。内部系统对接时,API 路径稳定比什么都重要。

5.2 并发能力瓶颈到底在哪

热搜里有"ai agent 怎么扛并发"这个问题,我的实测经验是:Agent 服务的并发瓶颈几乎从不在 FastAPI 层,而在模型推理服务和下游工具调用。FastAPI 的异步模型可以同时挂起成千上万个 IO 等待,但 vLLM 的并发能力是有上限的,内部系统的 API 服务响应速度也是有限的。

我做过一轮并发压测。用 200 个并发请求同时打 Agent 服务,每个请求内部有一次模型调用和一次工具调用。在 vLLM 配置max-num-seqs=32的情况下,系统表现是:约 30% 的请求在模型排队队列中等待,平均响应时间从单请求的 1.8 秒拉高到 6 秒左右。整体没有报错,但响应时间明显劣化。这说明系统的"扛并发"能力,本质上取决于下游服务的吞吐上限,而不是网关层。

为了让 Agent 服务能撑住更高的并发,我做了三层优化:

  1. 请求合并(Batching):vLLM 支持 Continuous Batching,它会把同时到达的请求自动合并成一个批次进行推理,这个机制本身就能极大提升吞吐。关键是别人为限制max-num-seqs,在显存允许的情况下可以适度调大,实测 32 是一个稳妥值。

  2. 异步工具调用:如果 Agent 一次要调多个相互独立的工具,不要串行等待。在 LangGraph 的 tool 节点里,我用了asyncio.gather并发执行工具函数。比如某个任务需要同时查询销售数据和库存数据,整体耗时从两次串行的 2 秒降到了并行的 1.1 秒。

  3. 限流保护:给自己留一条退路。对普通用户限流在每分钟 30 个请求,对内部系统保留 50 个请求的配额。限流不是为了限制用户,而是为了防止突发流量把模型推理服务打爆。

下面做一个简单的对比:

优化策略优化前优化后效果
vLLM Batching单请求推理多请求共享批处理吞吐提升约 2.5 倍
工具并发调用串行等待asyncio.gather多工具场景耗时降 40%
限流保护无限制配额控制避免雪崩式失败

5.3 隔离内网里的鉴权怎么设计

外网服务可以直接用 OAuth2、JWT 等标准方案,但内网环境下,很多时候没有现成的统一认证中心。我的做法是:给每个接入系统分配一个 API Token,服务端用中间件校验。代码比较直接:

from fastapi.security import HTTPBearer, HTTPAuthorizationCredentials security = HTTPBearer() @app.middleware("http") async def check_token(request: Request, call_next): if request.url.path in ["/docs", "/openapi.json"]: return await call_next(request) token = request.headers.get("Authorization", "").replace("Bearer ", "") if not validate_token(token): return JSONResponse(status_code=401, content={"detail": "Invalid token"}) return await call_next(request)

Token 存在哪?我用 PostgreSQL 的一张表,字段包括app_id、token_hash、expires_at。每次校验查库即可。内网环境不需要太复杂的密钥管理,但一定要做好过期轮换。

5.4 超时与重试:网络抖动不可怕,可怕的是没有兜底

隔离内网环境里,网络抖动、服务重启、交换机链路不稳定,都是家常便饭。我早期因为没做超时控制,经常出现接口一直挂起、占用连接资源的情况。后来统一规定了三层超时:

  • FastAPI 层:请求超时 60 秒
  • 工具调用层:每次工具 API 调用超时 10 秒
  • 模型推理层:单次模型调用超时 30 秒

超时之后统一走重试逻辑,重试次数 2 次,两次之间间隔 1 秒。如果重试还失败,返回一个标准化的错误响应,告诉调用方具体失败在哪一步。这个设计习惯是:任何一次外部依赖的调用,都必须假设它可能失败,且必须有明确的失败表现。

6. 内网部署后的避坑清单:我踩过的六个坑,你直接绕开

整个项目实测部署后,我整理了一份避坑清单,都是真实遇到过的,不是网上抄来的。

坑 1:模型文件不全,启动 10 分钟才发现报错。解决方案是提前做find /data/models -name "*" | wc -l确认文件数量,再和下载时的文件数对比。少文件通常在解压阶段才暴露,很浪费排查时间。

坑 2:PostgreSQL 连接数爆掉。LangGraph 的 Checkpoint 每次写入都会占用一个连接,高并发下,连接池默认配置很容易被打满。解决方案是设置pool_size=20和max_overflow=10,并调大数据库的最大连接数。

坑 3:工具函数里用了外网地址。内网环境没有 DNS 解析外网域名的能力。我的工具层里曾经有一个日志上报函数,内部硬编码了一个公网地址,导致工具调用永远失败。排查时发现这个工具根本没被模型调用过,但其他工具偶尔会挂。后来统一自查了一遍所有工具代码,凡是要外网访问的通通改走内网代理。

坑 4:Embedding 模型被误杀。某次重启机器后,Embedding 服务没有配置开机自启,结果文档检索工具全部报错,但模型推理服务是正常的。这种"部分依赖挂掉"的场景,比整体挂掉更隐蔽。我增加了健康检查接口,定期探测所有服务的存活状态,并接入告警。

坑 5:上下文太长导致模型截断。内网业务文档动辄几万字,超出模型 max-model-len 后,后面的内容会被静默截断,Agent 可能会忽略关键信息。我的解法是在接入 Agent 之前,先做一个预处理:把长文档拆分成多个分段,再按需检索后拼成上下文。这个方案既省 token,又不会漏关键信息。

坑 6:人工审核节点卡住整个流程。如果某类请求设置了需要人工确认,但审核界面没有访问入口,任务会永远卡在等待状态。我在系统里加了一个"强制超时"逻辑,如果审核请求在 5 分钟内没有响应,自动标记为拒绝并通知管理员。这让任务不会无限期占用连接资源。

我个人的深刻体会是,隔离内网里的 AI Agent 工程,最大的敌人不是模型能力不够,而是"可用性设计"的缺失。外网环境下,很多问题可以被默认可用的公共组件掩盖,比如云上的日志服务自动重试、公共模型 API 的高可用集群。一旦搬到内网,所有不可用性都会暴露出来。所以,做隔离内网 Agent 工程,请把 50% 的精力花在容错和处理不可用上,剩下的才去优化功能效果。

7. 这套方案还能怎么扩展

文章写到这,主体内容基本讲完了。最后聊一点后续的扩展方向,供大家参考。

第一,如果要做"AI Agent 中台",也就是多个业务线共享同一套 Agent 能力,可以考虑把工具注册表改造成一个可配置的管理平台,让各个业务方通过界面注册自己的工具和提示词,而不是改代码发版。这样能极大降低后续维护成本。

第二,关于会话记忆。目前我用 PostgreSQL 保存了会话状态,但如果未来承载更多用户,建议引入专门的向量检索来做长期记忆——把聊天历史中的关键信息转成向量存起来,新会话开始时优先找回相关记忆,Agent 的连续对话体验会好很多。

第三,多模型路由。内网环境里模型资源是有限的,可以做一个路由层:简单问题走 7B 小模型,复杂问题走 14B 大模型,甚至可以让模型先自评任务复杂度再决定走哪条通道。我们实测做了一个简化版,整体资源消耗降低了约 30%,同时高难度问题的准确率没有下降。

隔离内网不等于落后,它其实逼着你把工程上的细节想明白。希望这篇实战记录能给你一些参考,少走几步我当时走过的弯路。

需要专业的网站建设服务?

联系我们获取免费的网站建设咨询和方案报价,让我们帮助您实现业务目标

立即咨询