☰
混合模型与智能体架构工程实战:四层编排体系与安全策略落地
2026/10/1 12:29:08 网站建设 项目流程

这篇接着我们内部技术连载的第5篇讲实战。项目代号叫55873,我们的做法用一句话能说清:用6+1+3混合模型打底,用四层智能体架构把能力串起来,再靠安全策略编排守住所有入口。这套系统已经跑了大半年,从最开始只有几十个测试用户,到现在支撑日常问答、垂直领域知识库、代码生成和一批自动化任务,整体稳定性和效果都明显好于只挂一个单一大模型的时候。适合正在搭AI应用架构、搞智能体平台、或者准备把多个大模型接进业务系统的朋友参考。我不打算从大模型原理讲起,重点聊这套模型体系为什么这么拆、四层智能体架构怎么落地、安全策略编排到底编什么,以及我们踩过的坑。

1. 为什么是"6+1+3":混合模型体系的选型逻辑

1.1 混合模型不是堆API,而是给每个模型定角色

很多人一听到混合模型,第一反应就是把市面上能买到的模型全接进来,谁好用谁。这是误区。模型接多了之后,路由、成本、上下文、输出风格全都会打架。一个请求进来,到底该让谁回答?答完以后风格跟上次不一样怎么办?每个模型都消耗成本,怎么控制?这些都是问题。

我们最后收敛出来的方案就是"6+1+3":6个通用基座模型、1个主力推理模型、3个专用辅助模型。

6个通用基座模型负责的是一般性任务,我按场景分了六类:日常对话、轻量问答、代码生成与解释、多模态识别(图片理解、文档OCR)、文本向量化、低延迟端侧任务。这6个角色不一定对应6个完全不同的模型品牌,有的是同一个开源模型的不同量化档位,有的是专门跑在本地的小模型,但它们的功能边界必须分开,不能混用。

1个主力推理模型是整套体系里最贵的那个,它不干杂活,只处理复杂规划、长文本总结、跨领域综合推理、任务拆解这类"需要动脑子的活"。比如用户提一个模糊需求,需要先把需求拆成几步再决定调哪个工具,这种场景就交给主力模型。

3个专用辅助模型是面向特定业务定制的。比如我们有一批医疗健康类问答,自己整理了数据集,用业务语料微调了一个领域问答小模型,回答的严谨度比通用模型明显靠谱。另外还有一个文档解析辅助模型、一个用于日志和对话摘要的轻量模型。这三个辅助模型的特点是:任务固定、输入输出格式稳定、成本低,可以放心让它们高频干活。

用团队来类比就很好懂:你不能让CEO去前台收快递,也不能让前台去定公司战略。模型也一样,每个模型都应该有清晰的岗位描述,混着用的结果就是系统不稳定、成本失控。

1.2 选模型时必须盯紧的三个指标

选这些模型的时候,我们主要看三个指标:上下文长度、成本、工具调用能力。

上下文长度不是越长越好,但你必须知道在实际业务里会用掉多少。我们自己总结了一个估算公式:

请求消耗tokens = 系统提示词 + 用户消息 + 历史摘要 + 工具返回内容

每次调用前按这个公式算一遍,然后预留20%的buffer。如果发现经常接近上限,优先做历史摘要压缩,而不是简单换一个更长窗口的模型。长窗口模型通常价格翻倍,而且窗口拉到特别长以后,模型对中间部分的注意力会明显下降,很多问题照样会漏。

成本这块是最容易被忽视的。我们做了个简单的对比:本地跑量化后的7B小模型,单次请求基本只耗电,成本接近零;中档云端API每千tokens有稳定但不高的小额费用;主力推理模型最贵,乘上调用量以后,一个月下来数字非常可观。所以我们的原则很明确:高成本模型只走关键路径,能给小模型做的绝对不上大模型。每个月都要看token分布,找出哪些请求在白白浪费高价token,然后调整路由规则。

工具调用能力是另一个坑。同样是模型,A厂商的函数调用格式很规范,B厂商同一个请求偶尔会漏参数或者自己编工具名。我们的测试方法很粗暴:写5条需要多工具配合的请求,比如"查一下明天的天气,再帮我写一条带温度的出行提醒",然后看模型能不能正确输出tool_call、参数类型是否完整、连续多轮调用会不会错乱。这个测试必须在选型阶段就做,否则后面编排层会非常痛苦,因为你根本分不清是模型问题还是编排逻辑问题。

1.3 本地模型与云端模型如何分工

标题里提到"本地AI模型部署""AI代理助手加本地模型",这块我们确实踩过不少坑,我把最终的分工方式说说。

我们的原则是"本地优先、云端保底"。

本地模型处理四类事情:隐私数据相关的请求、高频低难度任务、文本向量化、日志摘要。比如用户录入的个人健康档案,绝对不能发到外部API,这种请求全部走本地模型。另一个典型场景是向量化,知识库里的文档要被切块并转成向量,这是一天几十万次的高频动作,放云端成本太高,本地小模型又快又便宜。

云端模型处理的是复杂推理、长文创作、跨领域综合问答。这些任务本地小模型确实顶不住,强行让它做就是浪费时间还答错。

操作上,本地模型我们用Ollama跑量化版模型,并发上来以后接入vLLM统一管理显存。云端则接国内主流大模型API,没有特殊原因不要同时接太多厂商,接口格式的统一维护成本会很高。

还有一个必须提前设计的能力:降级。云端不可用的时候,自动把流量切到本地模型,效果差一点但服务不能全断。这个降级不是写个if else就完了,要定期演练,因为很多云端故障是慢慢劣化的,不是直接报错,你的健康检查必须能识别"响应变慢但没挂"的状态。

2. 四层智能体架构:编排层的分层设计

2.1 四层架构到底拆的是哪四层

智能体架构这个词现在被用得很泛,每个人理解都不一样。我先明确我们内部的定义:四层从上到下分别是接入网关层、模型路由层、智能体编排层、安全策略审计层。

接入网关层是唯一的对外入口,所有客户端只看到这一个接口。它做鉴权、限流、协议转换,跟业务没关系,纯粹是门卫。

模型路由层负责"这个请求到底该交给谁"。它维护所有模型的状态、健康检查、成本配额,一旦某台机器挂了就自动降级。这里的关键是,路由层不关心任务本身怎么执行,它只做请求到模型的分配。

智能体编排层是核心,负责任务拆解、状态管理、工具调用、上下文管理。用户说"帮我安排一个下周一的会议,顺便订一间能坐十个人的会议室",编排层要先拆成"查日历、找空闲会议室、创建会议邀请"三个步骤,依次执行,最终汇总结果。

安全策略审计层不直接处理业务,它是横切在所有层之上的。输入要过它、输出要过它、工具调用要过它、审计日志也归它管。把安全放到独立层,而不是塞进某个业务代码里,是为了让策略可以统一升级、统一灰度。

我经常用饭店后厨来类比:接入网关是迎宾,模型路由是传菜员,智能体编排是厨师长,安全策略审计是食品安全员。传菜员不管菜怎么做,厨师长不管顾客怎么进店,食品安全员也不替厨师做菜,但每个环节都要检查。

2.2 编排层里最关键的不是模型,是任务状态机

很多人以为智能体编排就是"调用模型、拿到返回、再调用下一个模型",这是把编排层想简单了。真正跑起来你会发现,最大的问题是任务执行到一半失败了怎么办。

比如一个任务拆成了5步,第3步调天气接口超时了,你是从第1步重跑,还是从第3步重试?重跑的话,前面已经花掉的token全浪费;不重跑的话,第3步的上下文要怎么接回来?这时候就需要状态机。

我们的AgentState状态定义大致是:

  • pending:请求刚进入编排层,还没开始处理
  • routing:正在选模型
  • planning:主力模型正在拆解任务
  • executing:正在执行某个步骤
  • waiting_tool:已经发出工具调用,等待返回
  • verifying:校验工具返回结果是否正常
  • done:全部完成
  • failed:失败,记录原因,可按策略重试

每个状态都对应明确的处理逻辑。比如waiting_tool状态如果超过30秒没有收到返回,就自动进入failed,然后根据重试次数决定是从当前步骤重试还是上报人工。

有了状态机以后,重试、取消、超时全都变成了状态迁移问题,代码复杂度反而降低。你不需要在每个工具调用后面写一堆异常处理逻辑,只需要在状态迁移的地方统一处理。

还有一个容易被忽视的点:断点续跑。状态机把每一步的上下文都持久化下来之后,如果进程重启了,还能从最后的状态继续执行,而不是整个任务作废。这个在长耗时任务里特别重要。

2.3 模型路由策略:怎么判断一个请求该交给谁

路由层看起来简单,实际跑起来要考虑的规则非常多。我们最后用的是"硬规则优先、意图分类兜底、成本配额限流"三阶段策略。

第一阶段是硬规则,不需要模型判断,直接通过规则命中。比如请求里包含员工ID、健康档案、财务数据,直接走本地模型;请求来自代码生成工具,走代码模型池;请求是批量导入的离线任务,走低成本模型。

第二阶段是意图分类。用一个轻量模型给请求打标签,常见标签有chat、qa、code、agent_task、domain_qa。这一步的目的是把不好用硬规则判断的模糊请求分到正确的模型池里。

第三阶段是成本配额。高成本模型给每个用户设置每天调用上限,超过之后自动降级到通用模型。这里要注意,降级不是简单换模型,而是要给用户一个提示,说明当前是精简模式,让用户对回答质量有预期,否则体验会很差。

我用Python伪代码描述一下核心逻辑:

def route_request(request, user_profile): # 1. 硬规则 if contains_sensitive_data(request): return "local_chat_v3" if request.source == "code_agent": return "cloud_code_model" if request.is_batch_task: return "local_chat_v3" # 2. 意图分类 intent = classify_intent(request.text) if intent == "domain_qa": return "domain_qa_model_v3" if intent == "complex_qa": return "cloud_reasoner" return "local_chat_v3" # 3. 成本配额检查 if user_profile.daily_high_cost_used >= LIMIT: return "local_chat_v3"

这个伪代码省略了健康检查和熔断逻辑,但思路已经很清楚。路由逻辑要跟模型适配器完全解耦,路由只返回模型ID,具体怎么调用由网关层处理。

3. 安全策略编排:不是拦一道,而是分层治理

3.1 安全策略要覆盖四个位置

很多人做安全,习惯在入口加一个关键词过滤器,感觉就万事大吉了。实际根本不够,因为风险点分散在整条链路的各个位置。我们把它分成四个位置:

输入侧:用户请求可能带恶意内容或提示注入,比如试图覆盖系统提示词、诱导模型输出违规内容。这里要做的有内容合规检查、敏感信息识别、提示注入特征检测。

模型侧:模型本身可能被诱导或者出现幻觉。我们能做的是模型白名单管理,禁止调用未经评审的模型;同时对系统提示词做强制注入,用户无法通过对话改写系统规则。

工具侧:工具调用是最容易出安全问题的地方。比如模型被诱导去调一个不该调的内部接口,或者参数里带了危险命令。所以要维护工具白名单,对工具参数做类型和范围校验。

输出侧:模型生成的回答可能泄露训练数据里的信息,或者包含敏感内容。输出过滤要检查泄露关键词、异常内容,同时把错误信息做脱敏,不能把系统报错原样返回给用户。

这四层的策略不是简单的叠加关系,而是各有侧重。我做一个表格方便看:

位置主要风险常用手段
输入侧提示注入、敏感信息上传关键词过滤、正则、模型分类
模型侧模型幻觉、非白名单模型被调用模型白名单、系统提示词加固
工具侧危险操作、参数越权工具白名单、参数校验
输出侧内容泄露、错误信息暴露输出过滤、脱敏、审计日志

3.2 策略编排:用"策略即代码"的方式来管

安全策略一定不能用一堆if/else硬编码在业务代码里。因为策略变化太频繁了,硬编码的方式改一次要发一次版,而且容易漏。我们用的是声明式策略配置,把规则写在YAML文件里,由统一的策略引擎加载执行。

举个例子:

policies: - id: input-sensitive-block stage: input action: block rules: - type: keyword keywords: ["可疑词A", "可疑词B"] - type: regex pattern: "(?i)特殊模式正则" - id: tool-allowlist stage: tool action: allow rules: - type: field_in field: tool_name values: ["search_kb", "calc", "weather_api", "send_notification"] - id: output-redact-idcard stage: output action: redact rules: - type: regex pattern: "\d{17}[\dXx]"

每个策略有四个关键字段:id、stage、action、rules。stage表示这个策略在哪个环节生效,action有三种:block(直接拦截)、redact(打码后放行)、needs_review(转人工复核)。

为什么非要区分block和redact?因为硬拦截很容易误伤。比如医疗问答里出现"高血压""糖尿病"这些正常词汇,如果走了校验名单,整条回答就被拦掉太可惜。分档之后,正常词可以走needs_review或者直接放行,只有确凿的违规内容才block。

策略引擎还应该支持灰度发布。不能一上线就全局生效,先放5%的流量观察命中率,确认没有大面积误伤再逐步推广。这里特别提醒:安全策略不是越严越好,是在防风险和不影响正常体验之间找平衡。

3.3 审计日志记什么才有价值

很多团队的审计日志只是记录"谁在什么时间调了什么API",然后就没有然后了。真正的审计日志要能被复用来复盘和调优。

我们每条请求都会记录这样一个JSON:

{ "trace_id": "55873-20250415-001", "stage": "execution", "input_summary": "用户咨询高血压饮食注意事项", "output_summary": "生成饮食建议,未触发工具", "model_used": "domain_qa_model_v3", "policies_hit": [ "input_sensitive_check:pass", "tool_allowlist:pass" ], "token_count": 231, "cost_cents": 12, "tool_calls": [] }

这里我要特别强调两个很多人不记录但特别重要的字段:policies_hit和cost_cents。policies_hit记录这条请求命中了哪些安全策略、结果是pass还是block,这样才能知道策略有没有过度拦截。cost_cents记录这条请求花了多少钱,没有这个字段,混合模型的成本治理就是空话。审计日志至少要保留90天,并建立trace_id索引,方便从"用户反馈某条回答有问题"倒查到完整的调用链。

4. 实操过程与核心环节实现

4.1 从零搭一个最小可用的模型调度网关

我自己复现过很多次,最快的一次只花了一个下午。技术栈很简单:Python + FastAPI + Redis + 一个模型适配器。

先是模型配置文件,统一描述每个模型的基本信息:

models: - id: local_chat_v3 provider: ollama model_name: qwen2.5-7b-instruct context_window: 32768 pricing: 0 roles: [chat, prefilter] - id: cloud_reasoner provider: cloud_api model_name: reasoner-pro context_window: 65536 pricing: high roles: [plan, complex_qa]

然后定义一个统一的模型适配器接口。所有provider都必须实现同样的方法,后续接新模型只改配置,不用改业务代码。

class ModelAdapter(ABC): @abstractmethod def chat(self, messages: list[dict], tools: list[dict]) -> dict: """返回 content, tool_calls, usage"""

网关收到请求后做三件事:鉴权、限流、调用ModelRouter选模型,然后转发到对应适配器。

@app.post("/v1/chat") async def chat_endpoint(request: ChatRequest): user = await authenticate(request) await rate_limit(user.id) model_id = router.route(request, user) adapter = get_adapter(model_id) result = await adapter.chat( messages=request.messages, tools=request.tools ) return build_response(result)

在这里,Redis用来做限流计数和模型健康状态缓存。健康检查要单独起一个定时任务,每30秒探测一次所有模型,如果某个模型连续3次失败,就把它标记为不可用,路由层自动绕开它。

4.2 写一个最小的智能体编排内核

编排层的核心是AgentState的维护和循环执行。先定义一个状态数据结构:

@dataclass class AgentState: status: str = "pending" task_plan: list[str] = field(default_factory=list) current_step: int = 0 context: list[dict] = field(default_factory=list) tool_results: dict = field(default_factory=dict)

执行循环的逻辑大概是:

async def run_agent(goal: str, tools: list[Tool], max_steps: int = 10): state = AgentState(status="planning") state.context.append({"role": "user", "content": goal}) while state.status not in ("done", "failed"): if state.status == "planning": plan = await reasoner_plan(goal, tools) state.task_plan = plan state.status = "executing" elif state.status == "executing": next_action = await reasoner_next_action(state.context, tools) if next_action["type"] == "finish": state.status = "done" elif next_action["type"] == "call_tool": state.status = "waiting_tool" tool_name = next_action["tool_name"] tool_args = next_action["args"] state.context.append({ "role": "tool_call", "tool_name": tool_name, "args": tool_args }) tool_result = await execute_tool(tool_name, tool_args) state.tool_results[tool_name] = tool_result state.context.append({ "role": "tool_result", "content": tool_result }) state.status = "executing" if state.current_step >= max_steps: state.status = "failed" break return state

这段代码看起来简单,但实际有一个关键点:每一步的工具返回内容都要拼到context里,并交给主力模型继续推理。没有这一步,模型就不知道工具执行结果是什么,无法做下一步决策。

工具注册表长这样,每个工具必须暴露名字、描述、输入参数schema,这样模型才知道有什么工具可用、怎么传参:

tools = [ Tool( name="search_kb", description="搜索知识库,参数query为问题文本", parameters={"query": str} ), Tool( name="calc", description="执行数学计算,参数expression为表达式", parameters={"expression": str} ) ]

关于MCP这类标准协议,我们目前是作为一个可选项来设计的。统一的工具协议肯定是大方向,但现阶段最大的成本不在于协议选型,而在于把工具的输入输出schema调稳定。如果你的工具返回格式天天变,再好的协议也帮不上忙。

4.3 把安全策略编排接入主链路

策略引擎本身不复杂,核心是一个check方法:

class PolicyEngine: def check(self, stage: str, payload: dict) -> PolicyResult: policies = self.load_active_policies(stage) for p in policies: result = p.evaluate(payload) if result.action == "block": return PolicyResult(allow=False, reason=p.id) if result.action == "redact": payload = p.redact(payload) return PolicyResult(allow=True, redacted=True, reason=p.id) return PolicyResult(allow=True)

关键是调用时机要齐全,我的建议是至少接四个点:网关收到请求后先做input check;路由之前再做一次敏感数据识别;每次工具调用前做tool check;输出返回客户端之前做output check。

这四个检查点缺一个都可能出问题。比如只做input check不做output check,模型自己生成的敏感信息就没人管;只做tool check不做input check,用户可能在请求里夹带危险指令,模型还没到工具调用阶段就已经被带偏了。

5. 常见问题与排查技巧实录

5.1 混合模型调度最容易踩的坑

我列一个我们确实遇到过的坑位表,基本上每个做混合模型调度的团队都会碰到:

现象原因解决办法
换模型后回答风格突变不同模型的语气和措辞习惯差异太大统一系统提示词风格模板,输出前置风格约束
上下文窗口不够导致截断忽略工具返回长度,token预估不准按预估公式留buffer,历史做摘要压缩
路由误判关键词规则覆盖不全,意图分类模型能力不足每天用线上日志采样回流标注,迭代分类模型
工具返回格式不兼容不同工具返回结构混用,适配层没做好统一工具返回schema,适配器负责转换
本地模型并发高导致OOM量化模型也占显存,并发没限制限制并发数,用vLLM做推理管理
云端模型劣化但没报错健康检查只看HTTP状态码,没看响应质量增加响应延迟和结果完整性检查

这里我特别想提一下"回答风格突变"这个坑,它不致命但很影响体验。用户第一句问天气,本地小模型回答很简洁;第二句问复杂问题,切到主力模型,回答突然变成一大段分析。这不是逻辑错误,但用户会觉得产品不稳定。我们的解决办法是把系统提示词里的风格描述做成标准模板,无论哪个模型进来,都要先看到同一段风格约束,同时输出层再做一次长度和格式的规整。

5.2 智能体调试的三板斧

智能体出问题的时候,最难的不是修代码,而是复现。因为每一步都是动态的,工具返回不确定,模型输出也不确定。我们自己总结了三板斧,基本上能解决八成问题:

第一板斧是trace_id贯穿全链路。从请求进网关开始生成trace_id,中间经过路由、编排、工具调用、安全策略,全程携带。出问题时直接按trace_id把整条记录拉出来看,比在日志里搜关键词高效得多。

第二板斧是mock工具。真实工具依赖第三方接口,返回不稳定,容易掩盖编排逻辑的问题。调试阶段把工具全部mock成固定返回,先用稳定数据把编排逻辑调通,再切回真实工具。我测下来发现,很多编排层的问题不是工具没调通,而是模型拿到不确定的工具返回后不知道怎么办,mock之后问题马上就暴露了。

第三板斧是请求重放。把线上某条出问题的请求连同当时的上下文完整保存下来,在测试环境重放一遍。重放时锁定模型版本和工具返回,尽量保证可复现。不要小看这一步,没有重放能力,你永远在猜问题到底出在哪一层。

5.3 安全策略误伤如何收敛

安全策略做得太松等于没有,做得太紧又天天误伤。我们遇到过几个真实的误伤案例,挺有代表性的。

第一个案例是关键词硬拦截把正常医疗问询拦了。用户问"高血压患者能不能吃某某药",系统命中关键词"某某药"直接block。后来我们把这类词从硬拦截挪到needs_review,让人工或模型二次判断,误伤立刻下降。第二个案例是输出脱敏正则把产品名里的数字序列当身份证号打码了。这个没有特别好的办法,只能维护一个脱敏白名单,把常见正常序列提前排除。第三个案例是限流把内部批量任务卡死。内部数据同步任务每天上午固定跑一次,结果触发了单用户限流,任务失败。这个把内部任务打了独立限流分组才解决。

要收敛误伤,只有一个原则我觉得值得反复强调:安全策略必须分级。block、redact、needs_review三档缺一不可。而且策略上线前要做历史样本回测,把过去一周的真实请求拿过来跑一遍,看看拦截率和误伤率各占多少。安全策略本质上也是需要持续迭代的产品,不是写一组规则就完事。

6. 最后,几个我觉得特别值得分享的体会

6.1 最容易挂的地方不是模型,是状态管理和工具调用

我最初做智能体架构的时候,以为系统最弱的地方是大模型能力。跑了大半年以后回头看,真正让系统不稳定的是两件事:任务执行到一半怎么恢复状态,工具调用结果怎么被模型正确理解。大模型本身的能力已经超出预期,反而是工程化的细节决定系统能不能扛住生产流量。如果你正在搭智能体,我建议先把状态机和工具注册表做扎实,再追求模型效果。

6.2 别一步到位做平台,先做最小闭环

这个坑我们踩过。一开始总想着把模型管理、路由、编排、安全、审计、可视化全做齐,结果半年都还没上线。后来痛定思痛,先做了一个最小闭环:一个网关、一个编排器、三类安全策略。三个星期跑通第一条真实业务链路,后面再往上面加能力。做系统架构跟做产品一样,先跑通主流程,再丰富细节,顺序反了,项目很容易烂尾。

6.3 成本控制要靠token分布来驱动

混合模型最大的优势是省钱,但前提是要有监控。我们每个月做一次token分布分析,按照模型维度看调用量和成本占比。一般在第二个月就会发现一批完全可以走本地模型的请求,被错误地路由到了高成本模型上。把这类请求的收入是自己的收益。降级策略不能只在脑子里想,要真的写进代码并定期演练,否则真出事的时候发现降级逻辑本身有bug,那就尴尬了。

这套东西目前的形态还远谈不上完美,但已经足够把我们的精力从"接模型"转移到"做业务场景"上了。后面我们还在做智能体评估集、更细粒度的策略回测,以及工具协议的统一收口,这些经验等有结论了再继续写。

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

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

立即咨询