早期微信AI机器人的架构很直接:收到消息调大模型,模型回答里要查订单就由后端调业务接口。业务复杂度上来后这种直连模式会全面失控——模型想换一家要改遍所有调用点、业务接口鉴权方式各不相同、模型调用成本没人管、敏感对话直接发给外部模型。新阶段的架构思路是引入两个网关:模型网关统一管理所有大模型调用,工具网关统一管理所有业务能力,业务编排层只跟两个网关对话,不再直连任何模型或系统。

一、模型网关——把大模型当成一个可治理的资源池

业务代码里散落着十几处直接调用大模型的代码,是多数团队的真实状态。模型网关把这些调用全部收口,对上提供统一的对话、分类、抽取、向量四类标准接口,对下管理多个模型提供方。它解决的不是"能不能调通",是四个治理问题。

路由与降级:不同任务按难度和成本路由——意图分类用小模型,复杂推理用旗舰模型,多模态理解用视觉模型;某个模型超时或限流时自动降级到备用模型,业务无感知。成本控制:按场景设token预算,高频场景强制走小模型和语义缓存——客户问重复问题的比例很高,相似问题的答案命中缓存直接返回,不调模型。数据安全:对话内容出门前做脱敏,手机号、身份证、订单号替换成占位符,模型返回后还原。观测:每次调用的模型、耗时、token、成本、失败原因统一记录,成本异常和效果回归都有据可查。

二、工具网关——业务能力的统一注册中心

工具网关是业务系统能力的唯一出口。ERP、CRM、工单、库存等系统的接口在这里注册成标准工具,每个工具声明名称、描述、参数模式、鉴权要求、风险级别。对上,Agent和业务编排通过统一协议调用工具,不用知道底层是哪个系统的什么接口;对下,网关处理各系统千差万别的鉴权、协议、数据格式。

工具网关的核心价值在三点。注册即用:新业务系统接入只需在网关注册工具描述和适配器,上层能力立刻可被Agent调用,不用改编排代码。统一管控:鉴权、限流、熔断、审计在网关层集中实现——前面讲业务系统对接时的治理措施天然有了落地点。能力编排:复杂动作("查客户最近订单并判断能否改地址")由网关把多个原子工具组合成复合工具对外暴露,Agent面对的是语义化能力而不是一堆零散接口。工具描述同时供两类消费者使用——人看的接口文档和模型看的function schema,一份定义两处生成,避免文档和实现脱节。

三、双网关协作——编排层只表达业务意图

有了两个网关,业务编排层变得很薄:它不关心用哪个模型、不关心订单数据存在哪个系统,只表达"理解这条消息、判断意图、需要什么数据、执行什么动作、如何回复"。模型调用走模型网关,数据获取和动作执行走工具网关,编排层专注业务流程本身。

这种架构还有一个关键收益是两侧独立演进:模型升级换代(更强模型发布、价格调整、国产化替换)只改模型网关配置;业务系统迁移(ERP换厂商、CRM版本升级)只改工具网关适配器——两侧任何变化都不波及业务编排。初期看起来多了一层,但系统超过三个模型调用点和两个业务系统后,省下的联调和改造成本远超网关建设成本。

两种直连问题对照

直连模型的问题

模型网关对策

直连业务系统的问题

工具网关对策

换模型改全部代码

统一接口+路由

鉴权协议各自不同

注册+适配器

单点故障无降级

多提供方自动切换

熔断限流各写一套

集中管控

成本失控

预算+语义缓存

新系统接入成本高

注册即用

敏感数据裸发

出门脱敏还原

Agent面对零散接口

原子工具复合编排

双网关架构实现

# 模型网关:统一入口
class ModelGateway:
    def chat(self, scene, messages, **kw):
        if self.cache_hit(messages):          # 语义缓存
            return self.cache.get(messages)
        plan = ROUTE_TABLE[scene]             # 场景路由
        for model in plan["fallback_chain"]:  # 降级链
            try:
                safe_msgs = self.mask(messages)  # 脱敏出门
                resp = model.invoke(safe_msgs,
                    timeout=plan["timeout"])
                self.budget.check(scene, resp.usage)
                result = self.unmask(resp.text)
                self.metrics.record(model, resp)
                self.cache.set(messages, result)
                return result
            except ModelError:
                continue                      # 尝试下一个模型
        raise AllModelsFailed()

# 工具网关:注册中心
class ToolGateway:
    def register(self, tool):
        self.tools[tool.name] = tool

    def call(self, name, params, caller):
        tool = self.tools[name]
        self.authz.check(caller, tool)        # 统一鉴权
        self.ratelimit.acquire(tool)
        with self.circuit(tool):              # 熔断
            raw = tool.adapter.invoke(params) # 协议适配
        result = self.normalize(raw, tool.schema)
        self.audit.record(name, caller, params, result)
        return result

    def compose(self, composite_name):
        """原子工具组合成复合能力"""
        def composed(params):
            order = self.call("query_order",
                {"order_no": params["order_no"]}, caller="agent")
            if order["status"] == "shipped":
                return {"editable": False, "reason": "已发货"}
            return {"editable": True, "order": order}
        return composed

# 编排层:只表达业务意图
def handle_customer(wxid, text):
    intent = model_gateway.chat("intent_classify", [text])
    if intent == "change_address":
        info = tool_gateway.call("order_with_address_editable",
            parse_order_no(text), caller="bot")
        reply = model_gateway.chat("reply_compose",
            [text, json.dumps(info)])
        eyun.send_text(wxid, reply)

落地建议

双网关不要等系统复杂了才补——第一个模型调用和第一个业务接口对接时就走网关,哪怕网关里当时只有一个模型、一个工具,后续扩展零迁移成本。模型网关优先做路由和语义缓存,成本收益最直接;工具网关优先做注册和鉴权,管控能力随接入系统增多逐步补齐。微信侧的消息收发由Eyun这类个人微信API平台承接,建议把它也注册为工具网关中的一组原子工具(发消息、打标签、拉群),与业务工具接受同一套治理,接口参数和返回码以Eyun平台的开发文档为准。

Logo

DAMO开发者矩阵,由阿里巴巴达摩院和中国互联网协会联合发起,致力于探讨最前沿的技术趋势与应用成果,搭建高质量的交流与分享平台,推动技术创新与产业应用链接,围绕“人工智能与新型计算”构建开放共享的开发者生态。

更多推荐