1. 项目缘起与整体架构设计1.1 为什么要在隔离内网里折腾 AI Agent先说清楚这个项目的背景。我所在的团队负责一套内部业务系统的智能化改造这套系统跑在完全隔离的内网环境里没有公网出口没有外部 API 调用通道所有数据不出机房。在这种条件下想引入 AI Agent 能力第一反应肯定是不可能——毕竟大家平时用的那些云端大模型服务、在线工具链一个都连不上。但需求是真实存在的。业务侧希望把一些重复性的工单处理、日志分析、配置核查工作交给 Agent 自动完成减少人工介入。这就逼着我们必须在隔离环境里把整套 AI Agent 工程跑通。经过差不多两个月的摸索和迭代我们最终落地了一套可用的方案核心思路可以概括为本地化模型推理 内网工具编排 审批闸门控制。这套方案解决的核心问题是在没有任何外部网络依赖的前提下让 AI Agent 能够调用内网系统里的工具、执行实际任务同时保证每一步操作都在可控、可审计的范围内。适合谁来参考我觉得三类人最有用一是做企业内部系统集成的工程师二是对 AI Agent 落地感兴趣但受限于网络环境的技术负责人三是想了解 Agent 工程化实践的开发者。1.2 整体架构的分层思路整个架构我分成了四层从下往上依次是模型推理层负责本地大模型的加载和推理这是整个 Agent 的大脑。我们选的是量化后的开源模型跑在内网的 GPU 服务器上通过本地推理服务暴露接口。Agent 编排层负责意图理解、任务拆解、工具调用决策。这一层是核心逻辑所在决定了 Agent 怎么思考和行动。工具接入层把内网系统里的各种能力封装成 Agent 可以调用的工具包括工单查询、配置读取、日志检索等。审批控制层所有涉及写操作的动作都必须经过这一层人工确认后才能执行。为什么要这样分层最直接的原因是解耦。模型可能会换工具可能会增删审批策略可能会调整如果全部揉在一起改一处就牵一发而动全身。分层之后每层通过明确定义的接口通信替换任何一层都不影响其他层。另一个考虑是安全边界。审批控制层独立出来意味着即使 Agent 编排层出了逻辑错误写操作也不会直接落到内网系统上。这层物理隔离式的设计在实际运行中救过我们好几次。1.3 技术选型背后的取舍选型这块我踩过不少坑这里把关键决策和理由说清楚。模型选择内网环境没法用在线大模型只能本地部署。我们试过几个不同参数量的开源模型最终选了一个中等参数量的版本量化到 4bit 后显存占用可控推理速度也能接受。参数量太小的模型在工具调用决策上准确率不够太大的又跑不动这个平衡点需要根据实际硬件来定。Agent 框架没有用那些依赖外部服务的框架而是基于本地可用的编排库自己搭了一套轻量级的调度逻辑。核心是一个状态机管理 Agent 从接收任务到完成任务的整个生命周期。工具协议这里参考了 MCPModel Context Protocol的设计思路定义了一套内网工具的描述规范。每个工具用 JSON Schema 描述输入输出Agent 根据 Schema 生成调用参数。这样做的好处是工具注册和发现标准化新增工具只需要按规范注册即可。通信方式内网各系统之间用 HTTP 消息队列的组合。同步查询走 HTTP异步任务走消息队列避免 Agent 在等待长时间任务时阻塞。提示选型时不要盲目追求最新最热的方案隔离内网环境下依赖越少、越自包含的方案越可靠。每引入一个外部依赖都要评估它在断网条件下能否正常工作。2. 核心细节解析与实操要点2.1 本地模型推理服务的搭建要点本地推理服务是整个 Agent 的基础设施这块搭不稳上面全是空中楼阁。我们用的是常见的本地推理框架把量化后的模型加载起来暴露一个兼容 OpenAI 接口格式的 HTTP 服务。为什么要兼容这个格式因为这样上层的 Agent 编排代码可以用统一的客户端来调用将来换模型也不用改上层代码。部署时有几个关键参数需要调上下文长度设成 8192 比较合适。太短了装不下工具描述和对话历史太长了显存吃不消。我们的工具描述加起来大概 2000 token留足对话空间后 8192 是个甜点值。并发数根据 GPU 显存来定。我们单卡 24G 显存量化模型占 14G 左右剩下的空间大概能支撑 4 到 6 路并发。超过这个数请求就会排队。推理温度工具调用场景下温度设低一点0.1 到 0.3 之间。温度高了 Agent 容易发挥创意生成不存在的工具名或者错误的参数格式。启动命令大概长这样python -m local_inference_server \ --model-path /models/quantized-model-4bit \ --context-length 8192 \ --max-concurrent 4 \ --temperature 0.2 \ --port 8000服务起来之后用 curl 测一下连通性curl http://localhost:8000/v1/chat/completions \ -H Content-Type: application/json \ -d {model:local,messages:[{role:user,content:你好}]}能正常返回就说明推理服务 OK 了。2.2 工具描述规范的设计细节工具接入层是 Agent 和实际系统之间的桥梁。每个工具都需要一份清晰的说明书告诉 Agent 这个工具是干什么的、需要什么参数、返回什么结果。我们定义的工具有这么几个字段字段说明是否必填name工具唯一标识英文小写下划线是description功能描述要写得让模型能理解是parameters参数定义JSON Schema 格式是category工具分类用于权限控制是requires_approval是否需要审批是description 这个字段特别关键。写得太简略模型不知道什么时候该调用写得太复杂又浪费上下文。我的经验是用一句话说清楚这个工具做什么再用一句话说明什么场景下用就够了。举个例子一个查询工单状态的工具描述name: query_ticket_status description: 根据工单ID查询工单的当前状态和处理进度。当用户询问某个工单的处理情况时使用此工具。 parameters: type: object properties: ticket_id: type: string description: 工单的唯一标识符 required: [ticket_id] category: read_only requires_approval: false注意category和requires_approval这两个字段。只读类工具不需要审批写操作类工具必须审批。这个区分在后面的审批机制里会详细说。2.3 审批机制的实现逻辑审批机制是隔离内网 Agent 的安全底线。核心原则很简单读操作放行写操作拦截。具体实现上Agent 在决定调用某个工具之前先检查这个工具的requires_approval字段。如果是 false直接执行如果是 true就把调用请求挂起生成一条审批记录推送到审批队列里。审批记录包含这些信息请求 ID工具名称和参数Agent 的推理依据为什么要调这个工具请求时间当前状态待审批/已通过/已拒绝审批人通过内网的审批界面看到这些信息后决定通过还是拒绝。通过的话Agent 继续执行拒绝的话Agent 收到拒绝信号重新规划任务。这里有个细节值得说审批超时怎么处理。我们设了 30 分钟的默认超时超时后自动拒绝Agent 收到超时通知后走降级逻辑。为什么不设自动通过因为写操作的风险不可控宁可让任务失败也不能让未审批的操作溜过去。注意审批机制一定要做成默认拒绝而不是默认通过。任何异常情况超时、审批人不在、系统故障都应该导向拒绝这是安全设计的基本原则。3. 实操过程与核心环节实现3.1 从零搭建 Agent 编排引擎编排引擎是整个项目最核心的部分我把它拆成了几个模块来实现。任务接收模块Agent 的入口接收来自业务系统的任务请求。请求格式统一为 JSON包含任务描述、上下文信息、优先级等字段。接收后生成一个任务 ID写入任务队列。意图理解模块拿到任务描述后调用本地模型进行意图识别。这一步的输出是结构化的任务类型和关键参数。比如用户说帮我看看工单 T12345 现在什么情况模型输出{intent: query_ticket, params: {ticket_id: T12345}}。规划模块根据意图和可用工具列表生成执行计划。简单任务一步就能完成复杂任务可能需要多步。规划模块的输出是一个步骤列表每个步骤指定用哪个工具、传什么参数。执行模块按计划逐步执行。每执行一步把结果反馈给模型让模型判断是否需要调整后续计划。这就是所谓的 ReAct 模式——推理、行动、观察、再推理。状态管理模块维护整个任务的生命周期状态包括当前步骤、已完成步骤、中间结果等。状态持久化到本地数据库Agent 重启后能恢复。代码结构大概是这样class AgentOrchestrator: def __init__(self, model_client, tool_registry, approval_service): self.model model_client self.tools tool_registry self.approval approval_service self.state_store StateStore() def run(self, task): task_id self.state_store.create_task(task) intent self.understand_intent(task) plan self.plan(intent) for step in plan: result self.execute_step(step, task_id) if result.needs_replan: plan self.replan(task_id, result) return self.state_store.get_result(task_id) def execute_step(self, step, task_id): tool self.tools.get(step.tool_name) if tool.requires_approval: approved self.approval.request(step, task_id) if not approved: return StepResult(successFalse, reasonapproval_denied) return tool.invoke(step.params)3.2 工具注册与发现的实现工具注册这块我们做了一个注册中心。每个工具是一个独立的 Python 类继承自基类BaseTool实现invoke方法。注册的时候把工具的元信息写入注册表Agent 启动时从注册表加载所有可用工具。class BaseTool: name description parameters {} category read_only requires_approval False def invoke(self, params): raise NotImplementedError class QueryTicketStatus(BaseTool): name query_ticket_status description 根据工单ID查询工单状态 parameters {...} category read_only requires_approval False def invoke(self, params): ticket_id params[ticket_id] return internal_api.get_ticket(ticket_id)注册中心维护一个字典key 是工具名value 是工具实例。Agent 在规划阶段把所有工具的名称和描述拼成一段文本作为系统提示的一部分传给模型。模型根据这段描述决定调用哪个工具。这里有个优化点工具数量多了之后全部塞进上下文会爆。我们的做法是按 category 分组根据任务类型只加载相关分类的工具。比如查询类任务只加载read_only分类的工具减少上下文占用。3.3 内网系统对接的实际操作对接内网系统这块每个系统的接口风格都不一样需要做适配。我们抽象了一个InternalAPIClient封装了 HTTP 请求、认证、重试、超时等通用逻辑。认证用的是内网统一的 token 机制token 从配置中心获取定期刷新。重试策略是失败后等 1 秒重试最多 3 次。超时设 10 秒超过就认为接口不可用。class InternalAPIClient: def __init__(self, base_url, token_provider): self.base_url base_url self.token_provider token_provider self.session requests.Session() def request(self, method, path, **kwargs): token self.token_provider.get_token() headers {Authorization: fBearer {token}} for attempt in range(3): try: resp self.session.request( method, f{self.base_url}{path}, headersheaders, timeout10, **kwargs ) resp.raise_for_status() return resp.json() except requests.RequestException: if attempt 2: raise time.sleep(1)对接过程中遇到的最大问题是接口文档不全。有些老系统的接口没有文档只能靠抓包和试错来摸清楚参数格式。我的建议是对接之前先花时间把接口摸透写一个简单的测试脚本验证每个接口的输入输出确认无误后再封装成工具。磨刀不误砍柴工这一步省不得。3.4 并发处理与性能调优Agent 要扛并发这块必须认真对待。我们的场景是多个业务系统同时提交任务高峰期大概每秒十几个请求。第一层优化是推理服务并发。前面说了单卡能支撑 4 到 6 路并发超过就排队。我们的做法是加了一个请求队列超出并发数的请求先入队等有空闲槽位再处理。队列用优先级队列重要任务优先。第二层优化是工具调用异步化。有些工具调用耗时较长比如日志检索如果同步等待会阻塞整个 Agent 流程。我们把这些工具改成异步调用Agent 发起调用后继续处理其他步骤等结果回来再合并。第三层优化是结果缓存。对于查询类工具相同参数的请求在短时间内可能重复出现。我们加了一层缓存key 是工具名加参数哈希TTL 设 60 秒。这样重复查询直接命中缓存减少内网系统压力。from functools import lru_cache import hashlib class CachedToolWrapper: def __init__(self, tool, ttl60): self.tool tool self.ttl ttl self.cache {} def invoke(self, params): key hashlib.md5( f{self.tool.name}:{json.dumps(params, sort_keysTrue)}.encode() ).hexdigest() now time.time() if key in self.cache: value, expire_at self.cache[key] if now expire_at: return value result self.tool.invoke(params) self.cache[key] (result, now self.ttl) return result实测下来加了这三层优化之后系统在高峰期能稳定处理每秒 20 个左右的请求平均响应时间在 3 秒以内。对于内网系统来说这个性能已经够用了。4. 常见问题与排查技巧实录4.1 模型输出格式错误的排查这是最常见的问题。模型有时候不按预期的 JSON 格式输出导致解析失败。表现是 Agent 报无法解析工具调用参数。排查思路分三步看原始输出把模型的原始返回打印出来看看是格式问题还是内容问题。常见的是模型在 JSON 外面包了一层 markdown 代码块标记或者加了额外的解释文字。检查提示词提示词里有没有明确要求只输出 JSON不要有其他内容。如果没有加上。如果有但模型还是不听考虑用 few-shot 示例强化。加解析容错写一个健壮的解析函数能处理常见的格式偏差。比如先尝试直接解析失败后尝试提取 JSON 子串再解析。def parse_tool_call(text): try: return json.loads(text) except json.JSONDecodeError: pass # 尝试提取 JSON 子串 start text.find({) end text.rfind(}) 1 if start 0 and end start: try: return json.loads(text[start:end]) except json.JSONDecodeError: pass raise ValueError(f无法解析工具调用: {text[:200]})4.2 工具调用死循环的处理Agent 有时候会陷入死循环反复调用同一个工具或者在一个步骤上卡住出不来。这个问题很隐蔽因为从日志上看 Agent 一直在工作但实际上没有进展。我们的解决方案是加一个步骤计数器。每个任务设一个最大步骤数默认 20 步。超过就强制终止返回任务过于复杂请人工处理。另外加了一个重复调用检测。如果连续 3 次调用同一个工具且参数相同就判定为死循环中断执行。class LoopDetector: def __init__(self, max_repeat3): self.history [] self.max_repeat max_repeat def check(self, tool_name, params): signature (tool_name, json.dumps(params, sort_keysTrue)) self.history.append(signature) if len(self.history) self.max_repeat: recent self.history[-self.max_repeat:] if len(set(recent)) 1: return True return False4.3 审批流程卡顿的应对审批流程卡顿通常有两个原因一是审批人不在二是审批通知没送达。针对第一个问题我们设了多级审批人。主审批人 5 分钟没响应自动转给备审批人。备审批人也没响应再转给上级。这样保证审批请求不会因为某个人不在而无限期挂起。针对第二个问题审批通知走的是内网消息队列理论上不会丢。但实际运行中发现如果消息队列积压严重通知可能会延迟。我们的做法是加了一个定时扫描每 30 秒扫一次待审批列表发现有超过 5 分钟还没被处理的重新推送一次通知。4.4 常见问题速查表问题现象可能原因排查方法解决方案模型输出无法解析格式偏差打印原始输出加解析容错强化提示词Agent 死循环规划逻辑缺陷看步骤日志加步骤上限和重复检测工具调用超时内网系统慢看接口响应时间加超时和重试异步化审批卡住审批人不在看审批队列多级审批人定时重推推理服务 OOM并发过高看显存占用降并发加请求队列上下文超限工具描述太多看 token 数按分类加载工具提示排查问题时日志是第一手资料。建议在 Agent 的每个关键节点都打上详细日志包括输入、输出、耗时、状态。出问题时能快速定位比盲目猜测高效得多。4.5 几个踩过的坑和独家经验坑一模型对工具名的理解偏差。我们有个工具叫get_config模型有时候会理解成获取所有配置然后不传参数就调用。后来把名字改成get_config_by_key并在描述里强调必须提供配置键名问题就解决了。工具命名要尽量具体避免歧义。坑二审批参数被篡改。早期版本里审批人看到的参数和实际执行的参数是分开传递的存在被篡改的风险。后来改成审批和执行用同一份参数快照审批通过后直接执行快照里的参数杜绝了中间篡改的可能。坑三内网系统接口限流。有个内网系统对接口调用频率有限制Agent 高频调用触发了限流导致后续请求全部失败。后来在工具层加了令牌桶限流器控制调用频率问题解决。坑四模型版本升级导致行为变化。换了一个新版本的模型后同样的提示词Agent 的行为模式变了之前调好的参数需要重新调。教训是模型版本升级要当作一次重大变更来对待充分测试后再上线。经验一提示词要版本化管理。提示词是 Agent 行为的关键影响因素每次修改都要记录版本和变更原因。我们后来把提示词存在数据库里支持热更新和回滚方便调试。经验二灰度发布很重要。新版本的 Agent 不要一次性全量上线先拿少量任务试跑观察一段时间没问题再扩大范围。我们吃过一次亏新版本有个边界条件没处理好全量上线后出了不少问题。经验三保留完整的执行轨迹。每个任务的完整执行轨迹包括模型输入输出、工具调用记录、审批记录都要持久化保存。这不仅是审计需要更是排查问题的依据。我们后来靠这些轨迹定位了好几个隐蔽的 bug。这套系统跑到现在差不多半年了整体稳定。日均处理任务量在几千条审批通过率大概 85%剩下 15% 被拒绝或超时。Agent 的自动化率大概 70%也就是说七成的任务不需要人工干预就能完成。这个数字还在慢慢提升主要靠持续优化提示词和补充工具。后续我打算在几个方向继续打磨一是引入更细粒度的权限控制不同来源的任务用不同的工具集二是优化规划模块支持更复杂的多步任务三是把审批机制做得更智能低风险操作可以走快速通道。这些等有进展了再另开一篇聊。