1. 项目缘起与整体设计思路1.1 为什么要在隔离内网里折腾 AI Agent先说清楚这个项目的背景。我所在的团队负责一套企业级内部系统的运维和二次开发这套系统跑在完全隔离的内网环境里——没有外网出口没有公网镜像源连 pip install 都得走内部私有仓库。过去两年我们一直靠人工处理日志分析、工单归类、配置巡检这些重复劳动效率瓶颈非常明显。去年开始我陆续在个人环境里用各种 AI Agent 工具做自动化实验效果确实惊艳。但问题来了这些工具几乎全部依赖公网 API、在线模型服务和云端工具市场。一旦搬到隔离内网全部歇菜。于是就有了这个项目——在完全隔离的内网环境下从零搭建一套可用的 AI Agent 工程体系。这个项目的核心目标很明确让 AI Agent 在内网里真正“下地干活”而不是停留在演示阶段。具体来说要解决三类问题模型本地化没有公网 API模型必须本地部署且要能在有限算力下跑起来工具链闭环Agent 需要调用内部系统的接口、读写数据库、执行脚本这些能力要通过 MCP Tools 和 Skills 机制封装工程可维护不是跑个 Demo 就完事要能扛住日常并发、支持多 Agent 协作、方便后续扩展适合谁来参考如果你也在内网环境里做 AI 落地或者正在评估 AI Agent 的企业级部署方案这篇内容应该能帮你少踩不少坑。如果你只是想在个人电脑上玩玩 Agent那内网部署的很多约束你可能遇不到但工具封装和工程化的思路同样有参考价值。1.2 整体架构选型与背后的取舍逻辑架构设计这块我前后推翻了三个方案最终落地的版本是经过实际压测和日常使用验证的。先说选型逻辑。模型层内网环境没有公网 API所以必须本地部署。我试过三种路线——纯 CPU 推理、单卡 GPU 推理、多卡分布式推理。最终选了单卡 GPU 方案原因是我们的内网服务器只有一张 A100 80G多卡方案没有硬件基础纯 CPU 推理延迟太高单次响应超过 30 秒完全没法用于交互式场景。模型选的是 Qwen2.5-32B-Instruct 的量化版本用 GPTQ 4bit 量化后显存占用约 20G留出足够空间给 KV Cache 和并发请求。Agent 框架层这是最纠结的部分。我对比了 LangChain、LangGraph、Spring AI、以及基于 Rust 的一些轻量框架。最终选了LangGraph 自研调度层的组合。原因如下框架优势内网适配问题LangChain生态丰富工具多依赖大量在线组件内网裁剪工作量大LangGraph状态机清晰支持循环和分支需要自己补全持久化和并发控制Spring AIJava 生态友好对 Python 内部系统的适配成本高Rust 轻量框架性能好依赖少生态不成熟Skills 封装工作量大LangGraph 的核心优势在于它的状态图模型——每个 Agent 的行为可以拆解成节点和边节点之间的流转条件可以精确控制。这对于内网环境特别重要因为内网系统的接口调用往往有严格的顺序依赖和权限校验不能像公网 Agent 那样随意并发。工具层这是内网 Agent 工程和公网方案差异最大的地方。公网 Agent 可以直接调用各种在线 API内网 Agent 必须通过MCP Tools和Skills两层封装来对接内部系统。MCP Tools 负责底层能力暴露比如数据库查询、文件读写、HTTP 请求Skills 负责业务逻辑编排比如“生成巡检报告”这个 Skill 会依次调用日志查询 Tool、配置比对 Tool、报告生成 Tool。并发层内网系统的并发能力有限Agent 不能无限制地发起请求。我在调度层加了一个令牌桶限流器每个内部系统接口都有独立的令牌桶Agent 调用前先申请令牌。这个设计后面会详细讲。2. 核心细节解析与实操要点2.1 本地模型部署的关键参数与踩坑记录模型部署这块我踩的坑最多。先说结论内网部署模型量化方式和推理框架的选择比模型本身更重要。我们用的是 vLLM 作为推理框架原因是它支持 PagedAttention并发吞吐比 HuggingFace Transformers 原生推理高 3-5 倍。但 vLLM 在内网部署有几个坑坑一模型下载。内网没有 HuggingFace 访问权限模型文件必须提前下载好再拷贝进去。我建议用huggingface-cli download在公网环境下载完整模型目录然后通过内部文件传输通道拷贝。注意要下载完整的模型文件包括 tokenizer、config、generation_config 等缺一个都跑不起来。坑二CUDA 版本匹配。vLLM 对 CUDA 版本很敏感。我们内网服务器的驱动版本是 535支持 CUDA 12.2但 vLLM 最新版要求 CUDA 12.4。解决方案是装 vLLM 0.4.2 版本这个版本对 CUDA 12.2 兼容良好。这里提醒一句内网环境不要追求最新版本要追求最稳版本。坑三显存分配。A100 80G 看着很大但实际部署时要注意几个参数python -m vllm.entrypoints.openai.api_server \ --model /path/to/qwen2.5-32b-gptq \ --quantization gptq \ --dtype float16 \ --max-model-len 8192 \ --gpu-memory-utilization 0.85 \ --max-num-seqs 16 \ --port 8000关键参数解释--gpu-memory-utilization 0.85留 15% 显存给系统和其他进程设太高容易 OOM--max-num-seqs 16同时处理的最大请求数内网场景 16 足够设太高会导致单个请求延迟飙升--max-model-len 8192上下文长度32B 模型在 4bit 量化下8192 长度大约占 20G 显存实测下来这个配置在 A100 上单次推理延迟约 1.5-2 秒输入 500 token输出 200 token并发 8 个请求时延迟约 4-5 秒。对于内网工单处理场景这个性能完全够用。2.2 MCP Tools 封装让 Agent 安全地调用内部系统MCP Tools 是 Agent 和内部系统之间的桥梁。内网环境里Agent 不能直接访问数据库或执行 shell 命令必须通过 Tools 层做一层封装。这层封装不仅要暴露能力还要做权限控制、参数校验、审计日志。我封装了四类核心 Tools第一类数据库查询 Tool。内部系统用的是 PostgreSQLAgent 需要查询工单表、日志表、配置表。直接暴露 SQL 执行能力太危险我的做法是预定义查询模板from mcp.server import Server from mcp.types import Tool, TextContent server Server(internal-db-tools) server.tool() async def query_ticket_by_status(status: str, limit: int 50) - list[dict]: 根据状态查询工单status 可选值pending, processing, resolved allowed_status [pending, processing, resolved] if status not in allowed_status: raise ValueError(fstatus must be one of {allowed_status}) if limit 200: limit 200 # 实际查询逻辑 results await db.fetch( SELECT id, title, status, created_at FROM tickets WHERE status $1 LIMIT $2, status, limit ) return [dict(r) for r in results]这种设计的好处是Agent 只能执行预定义的查询不能拼接任意 SQL。每个 Tool 都有明确的参数约束和返回格式Agent 调用时不容易出错。第二类文件操作 Tool。Agent 需要读取日志文件、写入报告文件。内网环境里文件操作必须限制在特定目录下import os from pathlib import Path ALLOWED_BASE Path(/data/agent_workspace) server.tool() async def read_log_file(relative_path: str, max_lines: int 500) - str: 读取日志文件路径相对于 /data/agent_workspace full_path (ALLOWED_BASE / relative_path).resolve() if not str(full_path).startswith(str(ALLOWED_BASE)): raise PermissionError(Access denied: path outside workspace) if not full_path.exists(): raise FileNotFoundError(fFile not found: {relative_path}) with open(full_path, r) as f: lines f.readlines()[-max_lines:] return .join(lines)这里的关键是路径穿越防护——用resolve()解析真实路径后检查是否在允许目录内防止 Agent 通过../../etc/passwd这种方式越权访问。第三类HTTP 请求 Tool。内部系统有很多 REST 接口Agent 需要调用它们。但内网环境不能随便发请求必须限制目标地址import httpx from urllib.parse import urlparse ALLOWED_HOSTS [internal-api.company.local, monitor.company.local] server.tool() async def call_internal_api(url: str, method: str GET, body: dict None) - dict: 调用内部 API仅允许访问白名单内的主机 parsed urlparse(url) if parsed.hostname not in ALLOWED_HOSTS: raise PermissionError(fHost {parsed.hostname} not in allowlist) async with httpx.AsyncClient(timeout30) as client: if method.upper() GET: resp await client.get(url) elif method.upper() POST: resp await client.post(url, jsonbody) else: raise ValueError(fMethod {method} not supported) return resp.json()第四类脚本执行 Tool。有些运维操作需要执行 shell 脚本比如重启服务、清理临时文件。这类 Tool 风险最高我的做法是只允许执行预定义的脚本不允许 Agent 传入任意命令ALLOWED_SCRIPTS { restart_service: /opt/scripts/restart_service.sh, clean_temp: /opt/scripts/clean_temp.sh, check_disk: /opt/scripts/check_disk.sh, } server.tool() async def run_script(script_name: str, args: list[str] None) - str: 执行预定义脚本script_name 必须是 ALLOWED_SCRIPTS 中的键 if script_name not in ALLOWED_SCRIPTS: raise ValueError(fScript {script_name} not allowed) script_path ALLOWED_SCRIPTS[script_name] # 参数校验只允许字母数字和连字符 if args: for arg in args: if not all(c.isalnum() or c in -_ for c in arg): raise ValueError(fInvalid argument: {arg}) result subprocess.run( [script_path] (args or []), capture_outputTrue, textTrue, timeout60 ) return result.stdout result.stderr注意脚本执行 Tool 一定要设置超时否则一个卡死的脚本会把整个 Agent 拖垮。我设的是 60 秒超过就强制终止。2.3 Skills 编排把零散 Tool 组合成业务能力MCP Tools 是原子能力Skills 是业务能力。举个例子Agent 要完成“生成每日巡检报告”这个任务需要依次调用查询工单 Tool → 读取日志 Tool → 调用监控 API Tool → 写入报告文件 Tool。这四个 Tool 的调用顺序、参数传递、异常处理就是 Skill 要编排的内容。我用 LangGraph 的状态图来实现 Skill 编排。每个 Skill 是一个独立的状态图节点是 Tool 调用或 LLM 推理边是流转条件。下面是一个简化版的巡检报告 Skillfrom langgraph.graph import StateGraph, END from typing import TypedDict, Annotated import operator class InspectionState(TypedDict): date: str tickets: list logs: str metrics: dict report: str errors: Annotated[list, operator.add] def fetch_tickets(state: InspectionState): tickets await query_ticket_by_status(pending) return {tickets: tickets} def fetch_logs(state: InspectionState): logs await read_log_file(flogs/{state[date]}.log) return {logs: logs} def fetch_metrics(state: InspectionState): metrics await call_internal_api( http://monitor.company.local/api/metrics/daily ) return {metrics: metrics} def generate_report(state: InspectionState): prompt f根据以下数据生成巡检报告 待处理工单{state[tickets]} 日志摘要{state[logs][:2000]} 监控指标{state[metrics]} report await llm.generate(prompt) return {report: report} def save_report(state: InspectionState): await write_file(freports/{state[date]}_inspection.md, state[report]) return {} # 构建状态图 graph StateGraph(InspectionState) graph.add_node(fetch_tickets, fetch_tickets) graph.add_node(fetch_logs, fetch_logs) graph.add_node(fetch_metrics, fetch_metrics) graph.add_node(generate_report, generate_report) graph.add_node(save_report, save_report) graph.set_entry_point(fetch_tickets) graph.add_edge(fetch_tickets, fetch_logs) graph.add_edge(fetch_logs, fetch_metrics) graph.add_edge(fetch_metrics, generate_report) graph.add_edge(generate_report, save_report) graph.add_edge(save_report, END) inspection_skill graph.compile()这个 Skill 的设计要点状态传递每个节点返回的字典会合并到全局状态中后续节点可以读取前面节点的输出错误收集用Annotated[list, operator.add]定义 errors 字段任何节点出错都可以追加错误信息最后统一处理顺序控制用add_edge定义严格的执行顺序确保数据依赖正确实际运行中这个 Skill 完成一次巡检报告生成大约需要 8-12 秒包括 LLM 推理时间。如果某个 Tool 调用失败状态图会记录错误并继续执行后续节点最后在报告里标注哪些数据获取失败。3. 实操过程与核心环节实现3.1 从零搭建内网 Agent 环境的完整步骤这一节我把整个搭建过程拆成可复现的步骤。假设你有一台内网服务器装了 Ubuntu 22.04有一张 GPU显存 ≥ 24G下面是完整流程。第一步准备 Python 环境。内网没有公网 pip 源需要提前配置内部私有源。如果内部没有私有源可以用离线安装包的方式# 在公网环境下载所有依赖 pip download -r requirements.txt -d ./offline_packages # 拷贝到内网服务器后安装 pip install --no-index --find-links./offline_packages -r requirements.txtrequirements.txt 的核心依赖vllm0.4.2 langgraph0.0.40 langchain-core0.2.0 mcp0.9.0 httpx0.27.0 asyncpg0.29.0第二步部署模型服务。把量化后的模型文件拷贝到/data/models/qwen2.5-32b-gptq然后启动 vLLM 服务nohup python -m vllm.entrypoints.openai.api_server \ --model /data/models/qwen2.5-32b-gptq \ --quantization gptq \ --dtype float16 \ --max-model-len 8192 \ --gpu-memory-utilization 0.85 \ --max-num-seqs 16 \ --port 8000 /var/log/vllm.log 21 启动后验证curl http://localhost:8000/v1/models如果返回模型列表说明服务正常。第三步部署 MCP Tools 服务。把前面写的 Tools 代码打包成独立服务用 MCP 协议暴露# mcp_server.py from mcp.server import Server import asyncio server Server(internal-tools) # 注册所有 Tool server.tool() async def query_ticket_by_status(...): ... server.tool() async def read_log_file(...): ... # 启动服务 async def main(): async with server.run_stdio() as streams: await server.run(streams[0], streams[1], server.create_initialization_options()) if __name__ __main__: asyncio.run(main())第四步配置 Agent 调度层。调度层负责接收任务、调用 LLM、执行 Skill、返回结果。核心是一个 FastAPI 服务from fastapi import FastAPI from pydantic import BaseModel app FastAPI() class TaskRequest(BaseModel): skill_name: str params: dict app.post(/execute) async def execute_task(req: TaskRequest): skill SKILL_REGISTRY.get(req.skill_name) if not skill: return {error: fSkill {req.skill_name} not found} result await skill.ainvoke(req.params) return {result: result}第五步并发控制。内网系统的接口并发能力有限我在调度层加了一个令牌桶限流器import asyncio import time class TokenBucket: def __init__(self, rate: float, capacity: int): self.rate rate # 每秒补充的令牌数 self.capacity capacity # 桶容量 self.tokens capacity self.last_refill time.monotonic() self.lock asyncio.Lock() async def acquire(self, tokens: int 1): async with self.lock: now time.monotonic() elapsed now - self.last_refill self.tokens min(self.capacity, self.tokens elapsed * self.rate) self.last_refill now if self.tokens tokens: wait_time (tokens - self.tokens) / self.rate await asyncio.sleep(wait_time) self.tokens 0 else: self.tokens - tokens # 每个内部系统接口一个独立的令牌桶 ticket_api_bucket TokenBucket(rate5, capacity10) # 每秒 5 个请求最多积压 10 个 monitor_api_bucket TokenBucket(rate2, capacity5)在 Tool 调用前先申请令牌server.tool() async def query_ticket_by_status(status: str, limit: int 50): await ticket_api_bucket.acquire() # 实际查询逻辑 ...这个设计实测下来很稳。之前没有限流的时候Agent 并发高的时候会把内部工单系统打挂加了令牌桶之后即使 Agent 侧有 50 个并发请求实际打到工单系统的请求也被限制在每秒 5 个以内。3.2 并发场景下的 Agent 调度策略内网 Agent 的并发场景和公网不太一样。公网 Agent 通常是“一个用户一个会话”并发压力来自用户数量。内网 Agent 更多是“批量任务”场景比如一次性处理 200 个工单或者定时巡检 50 台服务器。这种场景下简单的“来一个请求起一个 Agent”模式会出问题200 个工单同时触发 200 个 Agent 实例每个实例都要调用 LLM 和内部 API瞬间把资源打满。我的做法是任务队列 固定 Worker 池import asyncio from asyncio import Queue class AgentWorkerPool: def __init__(self, num_workers: int 4): self.queue Queue() self.num_workers num_workers self.workers [] async def start(self): for i in range(self.num_workers): worker asyncio.create_task(self._worker(fworker-{i})) self.workers.append(worker) async def _worker(self, name: str): while True: task await self.queue.get() try: result await self._execute_task(task) task[future].set_result(result) except Exception as e: task[future].set_exception(e) finally: self.queue.task_done() async def submit(self, skill_name: str, params: dict): future asyncio.Future() await self.queue.put({ skill_name: skill_name, params: params, future: future }) return await future async def _execute_task(self, task): skill SKILL_REGISTRY[task[skill_name]] return await skill.ainvoke(task[params])Worker 数量设为 4 是基于实测A100 上 vLLM 的max-num-seqs是 16但每个 Agent 任务可能多次调用 LLM实际并发 LLM 请求数会超过 16。设 4 个 Worker 可以保证 LLM 请求不会排队太久同时内部 API 调用也不会过于密集。实操心得Worker 数量不是越多越好。我试过 8 个 Worker结果 LLM 请求排队严重单个任务延迟从 10 秒涨到 30 秒。降到 4 个之后整体吞吐反而更高。3.3 一个完整案例自动化工单分类与派发前面讲了很多架构和代码这一节用一个完整案例串起来。需求是每天有大量工单进入系统需要 Agent 自动读取工单内容、分类、派发给对应的处理组。Skill 设计class TicketDispatchState(TypedDict): ticket_id: int ticket_content: str category: str assignee_group: str dispatch_result: str def fetch_ticket(state): ticket await query_ticket_by_id(state[ticket_id]) return {ticket_content: ticket[content]} def classify_ticket(state): prompt f对以下工单进行分类只返回类别名称 类别选项网络问题、数据库问题、应用异常、权限申请、其他 工单内容{state[ticket_content]} category await llm.generate(prompt) return {category: category.strip()} def determine_group(state): mapping { 网络问题: 网络组, 数据库问题: DBA组, 应用异常: 应用组, 权限申请: 安全组, 其他: 综合组 } return {assignee_group: mapping.get(state[category], 综合组)} def dispatch_ticket(state): result await call_internal_api( http://internal-api.company.local/api/tickets/dispatch, methodPOST, body{ ticket_id: state[ticket_id], group: state[assignee_group] } ) return {dispatch_result: result[status]}执行流程调度层收到 200 个工单 ID批量提交到 Worker 池。每个 Worker 执行这个 Skill完成分类和派发。实测 200 个工单处理完成约 3 分钟平均每个工单 0.9 秒。其中 LLM 分类耗时约 0.6 秒API 调用耗时约 0.3 秒。效果对比人工处理 200 个工单大约需要 2-3 小时包括阅读、判断、手动派发Agent 处理只需要 3 分钟而且分类准确率在 92% 左右我抽样了 50 个工单人工复核。错误主要集中在“应用异常”和“其他”的边界上后续通过优化 prompt 和增加 few-shot 示例准确率提升到了 96%。4. 常见问题与排查技巧实录4.1 内网 Agent 部署的典型故障速查表这一节整理我在实际运维中遇到的高频问题。内网环境的故障排查比公网麻烦得多因为很多在线诊断工具用不了只能靠日志和手动验证。故障现象可能原因排查方法解决方案vLLM 启动报 CUDA 错误CUDA 版本不匹配nvcc --version和python -c import torch; print(torch.version.cuda)降级 vLLM 或升级驱动Agent 响应超时LLM 请求排队查看 vLLM 日志中的running和pending请求数减少 Worker 数量或增加max-num-seqsTool 调用返回 403内部 API 权限不足检查 Agent 服务账号的权限配置在内部系统里给服务账号授权文件读写失败路径不在允许目录内检查ALLOWED_BASE配置和实际路径调整配置或移动文件并发时内部系统报警请求频率过高查看内部系统的 QPS 监控调低令牌桶的 rate 参数模型输出乱码tokenizer 不匹配检查模型目录下的 tokenizer 文件重新下载完整模型文件Skill 执行卡死某个 Tool 没有超时设置查看 Skill 执行日志定位卡在哪个节点给所有 Tool 加超时4.2 那些文档里不会写的避坑经验经验一模型量化不要贪心。我一开始用了 3bit 量化显存确实省了但模型输出质量下降明显分类任务准确率从 92% 掉到 78%。后来换成 4bit 量化显存多用了 5G但准确率恢复到 91%。内网场景下模型质量比显存节省更重要因为内网任务通常对准确性要求高不像公网聊天机器人可以容忍一定错误。经验二Tool 的返回格式要统一。我早期写的 Tool 有的返回 dict有的返回 list有的返回 str导致 Agent 解析时经常出错。后来统一成所有 Tool 返回 dict并且包含status和data两个字段return {status: success, data: [...]} # 或 return {status: error, message: ...}这样 Agent 处理返回值时逻辑统一不容易出错。经验三日志要打全。内网环境没有在线监控出问题只能翻日志。我在每个 Tool 调用前后都加了日志import logging logger logging.getLogger(agent.tools) server.tool() async def query_ticket_by_status(status: str, limit: int 50): logger.info(fquery_ticket_by_status called: status{status}, limit{limit}) start time.monotonic() try: result await db.fetch(...) elapsed time.monotonic() - start logger.info(fquery_ticket_by_status success: {len(result)} rows, {elapsed:.2f}s) return {status: success, data: [dict(r) for r in result]} except Exception as e: logger.error(fquery_ticket_by_status failed: {e}, exc_infoTrue) return {status: error, message: str(e)}这些日志在排查问题时非常有用。有一次 Agent 批量处理工单时突然变慢翻日志发现是某个 Tool 的数据库查询没有走索引全表扫描导致延迟飙升。加上索引后恢复正常。经验四Skill 要有降级策略。内网系统不是 100% 可用的有时候监控 API 会短暂不可用。如果 Skill 里某个 Tool 调用失败就整个任务失败体验很差。我的做法是关键路径 Tool 失败则终止非关键路径 Tool 失败则跳过并记录def fetch_metrics(state): try: metrics await call_internal_api(...) return {metrics: metrics} except Exception as e: # 非关键路径记录错误但继续执行 return {metrics: {}, errors: [fmetrics fetch failed: {e}]}这样即使监控数据获取失败巡检报告仍然能生成只是在报告里标注“监控数据缺失”。经验五定期清理 Agent 工作目录。Agent 运行久了会在工作目录里积累大量临时文件占满磁盘。我加了一个定时清理 Skill每天凌晨执行一次删除 7 天前的临时文件。这个 Skill 本身也是用 Agent 执行的算是“用 Agent 管理 Agent”。4.3 性能调优的实操参数记录最后分享一组实测的性能数据供参考。测试环境A100 80GQwen2.5-32B-GPTQvLLM 0.4.2。配置项值说明max-model-len8192上下文长度再大显存不够gpu-memory-utilization0.85留 15% 给系统max-num-seqs16并发请求上限Worker 数量4Agent 并发执行数令牌桶 rate工单 API5/s内部系统承受上限令牌桶 rate监控 API2/s监控系统承受上限Tool 超时60s防止卡死LLM 请求超时120s长文本生成需要更长时间单任务延迟工单分类派发约 0.9 秒。批量 200 任务总耗时约 3 分钟。LLM 单次推理延迟500 token 输入200 token 输出约 1.5-2 秒。并发 8 请求时单次延迟约 4-5 秒。这套参数在我们内网环境跑了三个月日常处理 500-800 个工单没有出现过系统崩溃或严重延迟。偶尔有内部系统维护导致 Tool 调用失败但 Skill 的降级策略保证了任务不会完全中断。我个人在实际操作中的体会是内网 AI Agent 工程最难的不是模型部署也不是框架选型而是如何在资源受限、系统脆弱的环境里找到平衡点。公网方案可以堆资源、可以重试、可以降级到云端内网方案每一步都要精打细算。但一旦跑通收益也是实实在在的——我们团队现在每天节省至少 4 个小时的重复劳动这些时间可以投入到更有价值的优化工作上。