AgentScope Java 2.0：从入门到进阶

第 20 章 渠道（Channel）

本章导读：同一个 agent 要接到网页、命令行和即时通讯工具上，每接一个平台都要重新回答同一组问题——这条消息来自哪个用户、属于哪个对话、两条消息同时到达谁先谁后、注册了多个 agent 时交给谁。本章讲 AgentScope 对这组问题的回答：应用代码与 agent 之间的 Gateway——会话管理、per-session 并发控制与多 agent 路由，以及把消息平台适配成这套路由模型的 Channel：SendOptions 的用法、流式事件与 SSE Controller、与暴露子 Agent 的双向对话、GatewayBootstrap 多 agent 接线、自定义 Channel 接口与内置 IM 适配器。读完本章，你可以把同一个 agent 稳定地接到 HTTP、SSE 或任意消息平台上。

flowchart LR A["你的需求
同一个 agent 接到网页
命令行与即时通讯工具"] --> B["渠道 Channel
应用与 agent 之间的 Gateway 接线"] B --> C["Gateway 路由
会话管理 · 并发控制 · 多 agent 路由"] B --> D["平台适配
解析消息来源 · 投递回复"] B --> E["流式事件与 SSE
回复流实时推给前端"] B --> F["多 agent 与自定义
GatewayBootstrap 接线 · 自定义 Channel"] C --> G["收益
同一个 agent 稳定接到
HTTP · SSE · 任意消息平台"] D --> G E --> G F --> G

本章导览图

它们做什么

先看没有这层抽象时会撞上的问题。用户双击发送按钮，两条相同的消息同时进入同一个对话，agent 和自己竞争；同一个用户从网页和命令行各发一条，应用对不上这是不是同一段对话；接入第二个平台时，会话与路由的逻辑要再写一遍。

Gateway 位于你的应用代码和 agent 之间，负责：

Channel 把消息平台（HTTP、WebSocket、Slack 等）适配成 Gateway 的路由模型。它负责解析消息来源、选定目标 agent、以及把回复投递回去。

大多数场景下你不需要直接和 Gateway 或 Channel 打交道——agent.channel(...) 会在后台自动完成所有接线。

快速开始

两层各管什么清楚了，直接看最短路径——一个 agent、一个 channel、三次发送：

HarnessAgent agent = HarnessAgent.builder()
    .name("assistant")
    .sysPrompt("你是一个有用的助手。")
    .model("dashscope:qwen-plus")
    .build();

// 绑定一个 ChatUI channel。
ChatUiChannel chat = agent.channel(ChatUiChannel.create());

// 发送消息。每个 userId 自动获得独立的 session。
Msg reply = chat.send(SendOptions.userId("user-1"), "你好！").block();

// 同一个用户，同一个 session——对话继续。
Msg followUp = chat.send(SendOptions.userId("user-1"), "再多说一些。").block();

// 不同用户，不同 session。
Msg otherUser = chat.send(SendOptions.userId("user-2"), "你好").block();

agent.channel(...) 会懒加载创建内部 gateway，注册当前 agent，并把 gateway 注入到 channel 中。调用之后 chat 就可以直接使用了。

SendOptions

SendOptions 告诉 channel 谁在说话、这属于哪个对话：

工厂方法 行为
SendOptions.userId("user-1") 每个用户一个 session（最常用）
SendOptions.of("user-1", "session-a") 指定 session——同一用户多个对话
SendOptions.userId("user-1").withAgentId("support") 在多 agent 场景下路由到指定 agent
SendOptions.userId("user-1").withAttribute("tenant", "acme") 给本轮 agent 调用附带字符串/类型化属性
SendOptions.userId("user-1").withRuntimeContext(rtc) 携带完整 RuntimeContext（例如 force-sync 开关） 
// 同一用户，两个独立对话
chat.send(SendOptions.of("user-1", "session-a"), "话题 A").block();
chat.send(SendOptions.of("user-1", "session-b"), "话题 B").block();

// 把应用侧上下文传入本轮 agent 调用
chat.send(
        SendOptions.userId("user-1")
                .withAttribute("tenant", "acme")
                .withRuntimeContext(
                        RuntimeContext.builder()
                                .put(AgentSpawnTool.CTX_FORCE_SYNC, true)
                                .build()),
        "排查这张工单")
        .block();

多模态 / 结构化消息

纯文本 send(String) 只是便捷方法。图片、音频或多段内容请传预构建的 Msg（或 List<Msg>）——每个 String 重载都有对应的 Msg / List<Msg> 版本（含 SendOptions 与 sendStream）：

Msg multimodal = Msg.builder()
        .role(MsgRole.USER)
        .content(
                TextBlock.builder().text("这张图里有什么？").build(),
                ImageBlock.builder()
                        .source(URLSource.builder().url("https://example.com/photo.png").build())
                        .build())
        .build();

chat.send(SendOptions.userId("user-1"), multimodal).block();
chat.send(SendOptions.userId("user-1"), List.of(multimodal)).block();
chat.send(multimodal).block(); // 单 session 模式

RuntimeContext 合并

Channel 路径会在 Gateway 内构建 RuntimeContext。调用方可通过 SendOptions / InboundMessage.runtimeContext() / ChannelRuntimeContextResolver 提供 caller base。合并顺序：

  1. 以 caller 上下文为起点（可为空）
  2. 若配置了 ChannelRuntimeContextResolver 且返回非 null，则 替换 caller base
  3. Gateway 再覆盖身份字段——sessionId（gw-…）、userId、msgContext、gateKey、outboundAddress——冲突时以 Gateway 为准

通过 GatewayBootstrap 接线：

GatewayBootstrap gw = GatewayBootstrap.builder()
        .agent("main", b -> b.name("assistant").model(model))
        .runtimeContextResolver(req ->
                RuntimeContext.builder(req.callerContext())
                        .put("tenant", resolveTenant(req))
                        .build())
        .build();
ChatUiChannel chat = gw.chatUiChannel();

也可以拿到 gateway 后调用 gateway.setRuntimeContextResolver(...)。不要把业务属性写进 MsgContext.extra——该 map 参与 session key 计算。

流式事件 + SSE

同步的 send 拿到的是完整回复，聊天界面要的是边生成边显示。sendStream() 返回 Flux<AgentEvent>，和 agent.streamEvents() 一样的细粒度事件流，但经过 gateway 路由并带有会话管理。

chat.sendStream(SendOptions.userId("user-1"), "北京今天天气怎么样？")
    .doOnNext(event -> {
        if (event instanceof TextBlockDeltaEvent delta) {
            System.out.print(delta.getDelta());
        } else if (event instanceof ToolCallStartEvent tc) {
            System.out.println("\n[tool] " + tc.getToolCallName());
        }
    })
    .blockLast();

Spring Boot SSE Controller

@GetMapping(value = "/chat", produces = MediaType.TEXT_EVENT_STREAM_VALUE)
public Flux<ServerSentEvent<String>> chat(@RequestParam String message,
                                          @RequestParam String userId,
                                          @RequestParam(required = false) String sessionId) {
    SendOptions options = sessionId != null
            ? SendOptions.of(userId, sessionId)
            : SendOptions.userId(userId);

    return chat.sendStream(options, message)
            .map(event -> {
                Map<String, Object> payload = new LinkedHashMap<>();
                payload.put("type", event.getType().name());
                payload.put("id", event.getId());
                if (event instanceof TextBlockDeltaEvent delta) {
                    payload.put("delta", delta.getDelta());
                } else if (event instanceof SubagentExposedEvent se) {
                    payload.put("subagentId", se.getSubagentId());
                    payload.put("agentId", se.getAgentId());
                    payload.put("label", se.getLabel());
                }
                return ServerSentEvent.<String>builder()
                        .data(objectMapper.writeValueAsString(payload))
                        .build();
            });
}

与暴露的子 Agent 对话

当 agent 通过 expose_to_user=true spawn 子 agent 时，gateway 会把这个子 agent 暴露为用户可直接寻址的入口。一个 SubagentExposedEvent 会出现在 sendStream() 的事件流中，携带 subagentId。

发现暴露的子 Agent

AtomicReference<String> subagentId = new AtomicReference<>();

chat.sendStream(SendOptions.userId("user-1"), "找一个研究员帮我调查 AI 趋势")
    .doOnNext(event -> {
        if (event instanceof SubagentExposedEvent se) {
            subagentId.set(se.getSubagentId());
            System.out.printf("子 Agent 已暴露: id=%s agent=%s label=%s%n",
                    se.getSubagentId(), se.getAgentId(), se.getLabel());
        }
        if (event instanceof TextBlockDeltaEvent delta) {
            System.out.print(delta.getDelta());
        }
    })
    .blockLast();

SubagentExposedEvent 字段：

字段 说明
subagentId 用于向该子 agent 发消息的句柄
agentId 子 agent 类型（如 "researcher"）
sessionId 子 agent 的 session id
label 可选的人类可读名称

向子 Agent 发消息

拿到 subagentId 之后，可以直接和子 agent 对话——完全绕过父 agent：

// 非流式
Msg reply = chat.sendToSubagent(subagentId, "重点关注 LLM agent").block();

// 流式
chat.sendToSubagentStream(subagentId, "重点关注 LLM agent")
    .doOnNext(event -> {
        if (event instanceof TextBlockDeltaEvent delta) {
            System.out.print(delta.getDelta());
        }
    })
    .blockLast();

带子 Agent 支持的 SSE

典型的 SSE controller 同时处理主 agent 和子 agent 消息：

@GetMapping(value = "/chat", produces = MediaType.TEXT_EVENT_STREAM_VALUE)
public Flux<ServerSentEvent<String>> chat(@RequestParam String userId,
                                          @RequestParam String message,
                                          @RequestParam(required = false) String subagentId) {
    Flux<AgentEvent> events;
    if (subagentId != null) {
        events = chat.sendToSubagentStream(subagentId, message);
    } else {
        events = chat.sendStream(SendOptions.userId(userId), message);
    }
    return events.map(event -> toSSE(event));
}

客户端监听 SUBAGENT_EXPOSED 事件来渲染新的对话标签页，后续请求时把 subagentId 传回来即可。

多 HarnessAgent 路由

如果有多个 HarnessAgent 实例，使用 GatewayBootstrap：

HarnessAgent salesAgent = HarnessAgent.builder()
    .name("sales").sysPrompt("你是一个销售助手。")
    .model("dashscope:qwen-plus").build();

HarnessAgent supportAgent = HarnessAgent.builder()
    .name("support").sysPrompt("你是一个客服 agent。")
    .model("dashscope:qwen-plus").build();

GatewayBootstrap gw = GatewayBootstrap.builder()
    .agent("sales", salesAgent)
    .agent("support", supportAgent)
    .mainAgent("sales")          // 没有指定 agent 时的默认
    .build();

ChatUiChannel chat = gw.chatUiChannel();

按 agentId 路由

使用 SendOptions.withAgentId() 把消息路由到指定 agent：

// 路由到 sales（默认 main agent）
chat.send(SendOptions.userId("user-1"), "有什么产品？").block();

// 显式路由到 support
chat.send(SendOptions.userId("user-1").withAgentId("support"), "账单问题").block();

GatewayBootstrap 下暴露子 Agent

要在 GatewayBootstrap 模式下启用 expose_to_user，需要把 gateway bridge 接到每个 agent 的子 agent 中间件上：

GatewayBootstrap gw = GatewayBootstrap.builder()
    .agent("main", mainAgent)
    .build();

// 接入 bridge，让 agent_spawn(expose_to_user=true) 生效。
SubagentGatewayBridge bridge = gw.gatewayBridge();
// 通过 setGatewayBridge() 传给 agent 的 SubagentsMiddleware。

使用 agent.channel(...) 时，这个接线会自动完成。

自定义 Channel

实现 Channel 接口来适配新的消息平台：

public class MySlackChannel implements Channel {
    @Override public String channelId() { return "slack"; }
    @Override public ChannelConfig config() { return myConfig; }
    @Override public void init(Gateway gateway) { this.gateway = gateway; }
    @Override public void start() { /* 连接 Slack */ }
    @Override public void stop() { /* 断开连接 */ }

    @Override
    public Mono<Msg> dispatch(InboundMessage message) {
        RouteResult route = router.resolveRoute(config(), message);
        return gateway.run(route.context(), message.messages(), route.outboundAddress());
    }

    // 可选：流式分发
    @Override
    public Flux<AgentEvent> dispatchStream(InboundMessage message) {
        RouteResult route = router.resolveRoute(config(), message);
        return gateway.runStream(route.context(), message.messages(), route.outboundAddress());
    }
}

通过 GatewayBootstrap 注册：

GatewayBootstrap gw = GatewayBootstrap.builder()
    .agent("main", agent)
    .channel(new MySlackChannel())
    .build();

gw.start();   // 调用所有 channel 的 init() + start()
// ...
gw.stop();    // 调用所有 channel 的 stop()

内置 Channel 适配器

AgentScope 提供了多个开箱即用的 Channel 适配器作为扩展模块：

各适配器的接入配置详见第 25 章（IM 渠道集成）。

本章小结

把同一个 agent 稳定接到多个入口，这一章的机制按接入顺序串起来：

  1. 分工：Gateway 管会话映射、per-session 并发与 agent 路由；Channel 负责把具体消息平台适配到这套模型，agent.channel(...) 自动完成接线；
  2. 说清谁在说话：SendOptions 用 userId、session、withAgentId、属性与完整 RuntimeContext 表达来源与去向；纯文本之外传预构建的 Msg，业务属性不要写进 MsgContext.extra；
  3. 流式输出：sendStream() 提供与 streamEvents 同粒度、经 gateway 路由的事件流，SSE Controller 把它映射成 ServerSentEvent；
  4. 子 Agent 直连：SubagentExposedEvent 携带 subagentId 句柄，sendToSubagent / sendToSubagentStream 让用户绕过父 agent 与子 agent 对话；
  5. 多 agent：GatewayBootstrap 注册多个实例并指定默认，withAgentId 显式路由；GatewayBootstrap 模式下 expose_to_user 需要接 gatewayBridge，单 agent 的 agent.channel(...) 则自动完成；
  6. 自定义与内置：实现 Channel 接口即可接入新平台；钉钉、飞书 / Lark、GitHub、GitLab、企业微信适配器开箱即用，接入配置在第 25 章。

下一站把这套接入方式落到具体的消息平台——第 25 章（IM 渠道集成）逐个讲五个内置适配器的认证、签名校验与消息解析。

验收清单——读到这里，你应该能：

相关文档