20 KiB

Raw Blame History

Agent-Invoke 流式会话设计

日期: 2026-08-12 状态: Draft 关联模块: server/service/a2a/, server/service/agent-tool/executors/agent-invoke/, app/components/agent/, app/composables/useAgentChat.ts

1. 背景与目标

1.1 当前状态

agent-invoke 工具允许一个 agent 通过 A2A 协议调用另一个 agent 处理子任务。当前实现存在以下问题：

sendTask（server/service/a2a/service.ts:128）使用 generateText 同步调用 LLM，无流式输出
A2A 调用产生的对话消息未保存到 agentMessages 表，调用结束后无痕迹
前端无法感知 A2A 调用过程，只在 tool call 完成后看到最终结果文本
a2aTaskSessions 表仅记录 task 元数据（callerContext、state），不记录对话消息

1.2 目标

流式展示：A2A 调用过程中，被调用 agent 的回复实时流式展示在前端侧边面板
数据持久化：A2A 调用产生的 user/assistant 消息保存到 agentMessages 表，记录调用来源信息
会话隔离：A2A 产生的会话不在前端会话列表展示（仅用于审计/日志）
只读面板：侧边面板只读展示，用户不能在面板中继续对话

1.3 非目标

不支持用户在侧边面板中与被调用 agent 继续对话
不修改 A2A 协议的 JSON-RPC 接口（仅改内部 service 层）
不实现 A2A 调用的 streaming SSE 公开端点（仅内部订阅）

2. 架构设计

2.1 数据流总览

调用方前端 ← SSE ← 调用方 streamText (chat-engine)
  └─ tool-call: agent-invoke (args: {agentSlug, input})
     └─ agentInvokeExecutor.execute()
        └─ invokeAgentViaA2A()
           └─ createTask() → 创建 session + a2aTaskSessions 记录
              ↓ 返回 taskId
           └─ sendTask() → streamText (被调用 agent)
              └─ onChunk → appendChunk(taskId, chunk)
              └─ onFinish → saveMessage (source='a2a')
           └─ 返回 { output, ok, taskId }
        └─ 返回 ToolResult (metadata.taskId)
     └─ 调用方 streamText 继续，tool-output-available

前端侧:
  tool-call (agent-invoke) 开始
    → useAgentChat 检测到 toolName=agent-invoke
    → 自动打开 AgentInvokePanel
    → AgentInvokePanel 通过 EventSource 订阅 /api/agents/a2a/stream?taskId=xxx
    ← SSE chunks → 实时展示被调用 agent 回复
  tool-output-available
    → 面板可关闭，展示完成状态

2.2 关键设计决策

决策 1：streamBuffer 以 taskId 为 key

现有 stream-buffer.ts 以 sessionId 为 key。A2A 场景下，被调用 agent 的 session 对前端不可见，但 taskId 是调用方已知的（通过 tool call metadata 传递）。

方案：streamBuffer 支持以 taskId 为 key 创建独立 buffer，与 session buffer 隔离。新增 createStreamBufferByKey(key, meta) / appendChunkByKey(key, chunk) / subscribeToBufferByKey(key, ...) 等函数，或直接复用现有函数但传入 a2a_${taskId} 作为 key。

决策 2：taskId 传递给前端

agentInvokeExecutor.execute() 是同步等待 invokeAgentViaA2A 返回的。但前端需要在 tool 执行期间就订阅 SSE。

方案：invokeAgentViaA2A 在 createTask 完成后、sendTask 开始前，通过调用方 streamBuffer 推送一个自定义 SSE 事件 a2a-task-started，包含 { taskId, calleeAgentSlug, calleeAgentName }。前端在收到 tool-call chunk（toolName=agent-invoke）后，等待后续的 a2a-task-started 事件获取 taskId，然后打开面板订阅。

替代方案（更简单）：executor 在 execute 开始时立即创建 task（先 createTask 拿到 taskId），将 taskId 作为 tool call 的 args 的一部分通过 streamText 的 tool-input-available 事件传递给前端。但这需要改 executor 接口。

最终选择：通过调用方 streamBuffer 推送 a2a-task-started 事件。原因：不改 executor 接口，复用现有 streamBuffer 机制。

决策 3：消息保存与来源记录

在 agentMessages 表新增字段记录来源：

字段	类型	默认值	说明
`source`	text	`'user'`	消息来源：`'user'`（用户直接对话）\| `'a2a'`（A2A 调用）
`a2aTaskId`	text	null	A2A task ID，仅 a2a 消息有值
`callerAgentId`	integer	null	调用方 agent ID，references agents.id
`callerSessionId`	text	null	调用方会话 ID

会话列表过滤：listSessions 查询时，排除只含 source='a2a' 消息的 session。实现方式：在 a2aTaskSessions 表新增 visibleInSidebar 字段（integer, default 0），listSessions 查询时 LEFT JOIN a2aTaskSessions 并过滤 visibleInSidebar = 0 的 session。

3. 详细设计

3.1 数据层变更

3.1.1 Schema 变更

文件: packages/drizzle-pkg/lib/schema/agent.ts — agentMessages 表新增字段：

source: text("source", { length: 10 }).notNull().default("user"),
a2aTaskId: text("a2a_task_id", { length: 64 }),
callerAgentId: integer("caller_agent_id").references(() => agents.id, { onDelete: "set null" }),
callerSessionId: text("caller_session_id"),

新增索引：idx_agent_messages_source ON (source)

文件: packages/drizzle-pkg/lib/schema/a2a.ts — a2aTaskSessions 表新增字段：

visibleInSidebar: integer("visible_in_sidebar").notNull().default(0),

3.1.2 Migration

使用 drizzle-kit 生成 migration：

bun run db:generate
bun run db:migrate

3.1.3 查询层变更

server/service/agent/session.ts — listSessions：

查询条件增加：排除 a2aTaskSessions.visibleInSidebar = 0 的 session。

-- 原查询
SELECT * FROM agent_sessions WHERE user_id = ? AND deleted_at IS NULL

-- 新查询
SELECT s.* FROM agent_sessions s
LEFT JOIN a2a_task_sessions a ON a.session_id = s.id
WHERE s.user_id = ? AND s.deleted_at IS NULL
  AND (a.visible_in_sidebar = 1 OR a.id IS NULL)

3.2 后端流式改造

3.2.1 `invokeAgentViaA2A` 改造

文件: server/service/a2a/client.ts

在 createTask 完成后、sendTask 开始前，通过调用方 streamBuffer 推送 a2a-task-started 事件。

需要新增参数 callerSessionId（调用方会话 ID），用于找到调用方的 streamBuffer 并推送事件。

export async function invokeAgentViaA2A(
  invocation: AgentInvocation,
  options?: {
    maxRecursionDepth?: number;
    timeoutMs?: number;
    maxOutputTokens?: number;
    callerSessionId?: string;  // 新增
  },
): Promise<AgentInvocationResult> {
  // ... createTask 后
  const task = await createTask(...);

  // 推送 a2a-task-started 事件到调用方 streamBuffer
  if (options?.callerSessionId) {
    const event = `data: ${JSON.stringify({
      type: "a2a-task-started",
      taskId: task.id,
      calleeAgentSlug: agentSlug,
      calleeAgentName: agent.name,
    })}\n\n`;
    appendChunk(options.callerSessionId, new TextEncoder().encode(event));
  }

  // sendTask 改为流式
  const completedTask = await sendTask(task.id, a2aContext);
  // ...
}

返回结果新增 taskId：

return { agentSlug, output, ok: true, taskId: task.id };

3.2.2 `sendTask` 改造

文件: server/service/a2a/service.ts

从 generateText 改为 streamText，流式过程中：

创建以 a2a_${taskId} 为 key 的 streamBuffer
onChunk 将文本增量编码为 SSE 格式写入 buffer
onFinish 保存 user/assistant 消息到 agentMessages（带来源字段）

export async function sendTask(taskId, context): Promise<A2ATask> {
  // ... 前置校验同原逻辑

  // 创建 A2A streamBuffer
  const a2aBufferKey = `a2a_${taskId}`;
  createStreamBuffer(a2aBufferKey, { modelId: resolved.dbId });

  // 保存 user 消息（source='a2a'）
  const userSortOrder = (await getMaxSortOrder(taskRow.sessionId)) + 1;
  await saveMessage({
    sessionId: taskRow.sessionId,
    role: "user",
    content: inputText,
    sortOrder: userSortOrder,
    // 新增字段
    source: "a2a",
    a2aTaskId: taskId,
    callerAgentId: context.callerAgentId,
    callerSessionId: context.callerSessionId,  // 需在 A2AInvocationContext 新增
  });

  const result = streamText({
    model: languageModel,
    system: agent.systemPrompt,
    prompt: inputText,
    tools,
    stopWhen: stepCountIs(maxSteps),
    maxOutputTokens,
    abortSignal: combinedSignal,
    onChunk: ({ chunk }) => {
      if (chunk.type === "text-delta") {
        const sseChunk = `data: ${JSON.stringify({
          type: "text-delta",
          textDelta: chunk.text,
        })}\n\n`;
        appendChunk(a2aBufferKey, new TextEncoder().encode(sseChunk));
      }
    },
    onFinish: async ({ text, usage }) => {
      // 保存 assistant 消息
      const assistantSortOrder = (await getMaxSortOrder(taskRow.sessionId)) + 1;
      await saveMessage({
        sessionId: taskRow.sessionId,
        role: "assistant",
        content: text ?? "",
        modelId: resolved.dbId,
        inputTokens: usage?.inputTokens ?? null,
        outputTokens: usage?.outputTokens ?? null,
        sortOrder: assistantSortOrder,
        source: "a2a",
        a2aTaskId: taskId,
        callerAgentId: context.callerAgentId,
        callerSessionId: context.callerSessionId,
      });
      markBufferDone(a2aBufferKey);
      await updateTaskState(taskId, "completed");
    },
  });

  // 等待流式完成
  await result.text;

  const updatedRow = await getTaskRow(taskId);
  return rowToTask(updatedRow!, textToA2AMessage(output, "agent"));
}

3.2.3 `A2AInvocationContext` 类型扩展

文件: server/service/a2a/types.ts

export interface A2AInvocationContext {
  // ... 现有字段
  callerSessionId?: string;  // 新增：调用方会话 ID
}

3.2.4 `agentInvokeExecutor` 改造

文件: server/service/agent-tool/executors/agent-invoke/agent-invoke.ts

execute 方法传入 callerSessionId：

async execute(
  input: unknown,
  config: AgentInvokeToolConfig,
  ctx: { userId: number | null; agentId?: number | null; recursionDepth?: number; sessionId?: string },
): Promise<ToolResult> {
  // ... 现有逻辑
  const result = await invokeAgentViaA2A(
    {
      agentSlug: data.agentSlug,
      input: data.input,
      context: {
        userId: ctx.userId,
        callerAgentId: ctx.agentId ?? null,
        recursionDepth: currentDepth + 1,
      },
    },
    {
      maxRecursionDepth: config.maxRecursionDepth,
      timeoutMs: config.timeoutMs,
      maxOutputTokens: config.maxOutputTokens,
      callerSessionId: ctx.sessionId,  // 新增
    },
  );

  return {
    success: true,
    data: output,
    metadata: {
      durationMs: Date.now() - start,
      agentSlug: data.agentSlug,
      recursionDepth: currentDepth + 1,
      taskId: result.taskId,  // 新增
    },
  };
}

已确认：executor 的 ctx（ToolContext）当前不包含 sessionId。需要扩展以下接口：

ToolContext（server/service/agent-tool/registry.ts:11）新增 sessionId?: string
executeAgentTool（server/service/agent-tool/index.ts:302）新增 sessionId?: string 参数，传入 ctx
getAgentToolsForChatByAgentId（server/service/agent-tool/index.ts:546）新增 sessionId?: string 参数，传递给内部 executeAgentTool 调用
chat-engine.ts 调用 getAgentToolsForChatByAgentId 时传入 sessionId（chat-engine.ts 已有 sessionId 变量，line 191）

3.2.5 新增 SSE 端点

文件: server/api/agents/a2a/stream.get.ts

export default defineEventHandler(async (event) => {
  const query = getQuery(event);
  const taskId = query.taskId as string;

  if (!taskId) {
    throw createError({ statusCode: 400, statusMessage: "缺少 taskId" });
  }

  // 鉴权：验证调用方有权访问该 task
  // （通过 a2aTaskSessions 查询 callerAgentId / callerSessionId 匹配当前用户）

  const a2aBufferKey = `a2a_${taskId}`;
  const buf = getStreamBuffer(a2aBufferKey);

  if (!buf) {
    return R.success({ active: false, reason: "no-buffer" });
  }

  // 复用 stream.get.ts 的 ReadableStream 模式
  const stream = new ReadableStream({
    start(controller) {
      for (const chunk of buf.chunks) {
        controller.enqueue(chunk);
      }
      if (buf.done) {
        controller.close();
        return;
      }
      const unsubscribe = subscribeToBuffer(a2aBufferKey, ...);
      event.node.req.on("close", () => { unsubscribe(); });
    },
  });

  return new Response(stream, {
    headers: {
      "Content-Type": "text/event-stream; charset=utf-8",
      "Cache-Control": "no-cache",
      "Connection": "keep-alive",
      "X-Stream-Resume": "true",
    },
  });
});

3.3 前端变更

3.3.1 新增 `AgentInvokePanel` 组件

文件: app/components/agent/AgentInvokePanel.vue

<script setup lang="ts">
const props = defineProps<{
  taskId: string;
  calleeAgentSlug: string;
  calleeAgentName: string;
}>();

const emit = defineEmits<{
  close: [];
}>();

const streamedText = ref("");
const isDone = ref(false);
const error = ref<string | null>(null);

let eventSource: EventSource | null = null;

watchEffect(() => {
  if (!props.taskId) return;

  eventSource = new EventSource(
    `/api/agents/a2a/stream?taskId=${encodeURIComponent(props.taskId)}`,
    { withCredentials: true }
  );

  eventSource.onmessage = (e) => {
    try {
      const chunk = JSON.parse(e.data);
      if (chunk.type === "text-delta") {
        streamedText.value += chunk.textDelta;
      } else if (chunk.type === "done") {
        isDone.value = true;
        eventSource?.close();
      } else if (chunk.type === "error") {
        error.value = chunk.message;
        eventSource?.close();
      }
    } catch {}
  };

  eventSource.onerror = () => {
    if (!isDone.value) error.value = "连接中断";
    eventSource?.close();
  };
});

onUnmounted(() => eventSource?.close());
</script>

<template>
  <div class="agent-invoke-panel">
    <div class="panel-header">
      <Icon name="lucide:bot" />
      <span>{{ calleeAgentName }}</span>
      <span class="status">{{ isDone ? '完成' : '生成中…' }}</span>
      <button @click="emit('close')">
        <Icon name="lucide:x" />
      </button>
    </div>
    <div class="panel-body">
      <AgentMarkdown :content="streamedText" v-if="streamedText" />
      <div v-else-if="error" class="error">{{ error }}</div>
      <div v-else class="loading">
        <Icon name="lucide:loader-circle" class="spin" />
        <span>等待回复…</span>
      </div>
    </div>
  </div>
</template>
</parameter>

3.3.2 `useAgentChat.ts` 改造

在处理 tool-call chunk 时，检测 toolName === 'agent-invoke'，自动打开面板。

新增响应式状态：

const invokePanelState = ref<{
  visible: boolean;
  taskId: string;
  calleeAgentSlug: string;
  calleeAgentName: string;
} | null>(null);

在 chunk 处理逻辑中：

case "tool-input-available": {
  // ... 现有逻辑
  
  // 检测 agent-invoke 工具调用
  if (chunk.toolName === "agent-invoke") {
    // 解析 args 获取 calleeAgentSlug
    const args = typeof chunk.args === "string" ? JSON.parse(chunk.args) : chunk.args;
    // 先展示面板（taskId 暂空，等待 a2a-task-started 事件）
    invokePanelState.value = {
      visible: true,
      taskId: "",
      calleeAgentSlug: args?.agentSlug ?? "",
      calleeAgentName: args?.agentSlug ?? "",
    };
  }
  break;
}

case "a2a-task-started": {
  // 更新面板的 taskId，开始 SSE 订阅
  if (invokePanelState.value) {
    invokePanelState.value.taskId = chunk.taskId;
    invokePanelState.value.calleeAgentName = chunk.calleeAgentName;
  }
  break;
}

case "tool-output-available": {
  // ... 现有逻辑
  // agent-invoke 完成后，面板标记为完成（但不自动关闭，用户手动关闭）
}

3.3.3 `AgentChatArea.vue` 改造

挂载 AgentInvokePanel：

<template>
  <!-- 现有内容 -->
  <AgentInvokePanel
    v-if="invokePanelState?.visible"
    :task-id="invokePanelState.taskId"
    :callee-agent-slug="invokePanelState.calleeAgentSlug"
    :callee-agent-name="invokePanelState.calleeAgentName"
    @close="invokePanelState = null"
  />
</template>

3.4 错误处理

场景	处理
A2A 调用超时	streamBuffer 推送 error 事件，面板展示错误，executor 返回失败
被调用 agent 不存在/不可调用	executor 返回失败（现有逻辑），面板不打开
SSE 连接中断	面板展示"连接中断"，用户可手动关闭
前端刷新	streamBuffer 支持 resume（复用现有 stream.get.ts 模式），但 A2A 场景下 taskId 丢失，面板不恢复（可接受）
递归深度超限	executor 返回失败（现有逻辑），面板不打开

3.5 测试策略

3.5.1 单元测试

sendTask 流式输出：mock streamText，验证 onChunk 写入 buffer、onFinish 保存消息
invokeAgentViaA2A 推送 a2a-task-started 事件：验证 appendChunk 被调用
listSessions 过滤 A2A 会话：插入 A2A 消息，验证会话列表不包含

3.5.2 E2E 测试

配置两个 agent（通用助手 + 编程助手），通用助手配置 agent-invoke 工具
发送消息触发 agent-invoke，验证侧边面板自动打开并展示流式回复
验证 A2A 消息保存到数据库（source='a2a'）
验证会话列表不展示 A2A 会话

4. 影响范围

4.1 修改文件清单

文件	变更类型	说明
`packages/drizzle-pkg/lib/schema/agent.ts`	修改	agentMessages 新增 source/a2aTaskId/callerAgentId/callerSessionId
`packages/drizzle-pkg/lib/schema/a2a.ts`	修改	a2aTaskSessions 新增 visibleInSidebar
`server/service/agent-tool/registry.ts`	修改	ToolContext 新增 sessionId 字段
`server/service/agent-tool/index.ts`	修改	executeAgentTool + getAgentToolsForChatByAgentId 新增 sessionId 参数
`server/service/a2a/types.ts`	修改	A2AInvocationContext 新增 callerSessionId
`server/service/a2a/client.ts`	修改	invokeAgentViaA2A 推送 a2a-task-started 事件，返回 taskId
`server/service/a2a/service.ts`	修改	sendTask 改为 streamText，保存消息
`server/service/agent-tool/executors/agent-invoke/agent-invoke.ts`	修改	execute 传入 callerSessionId，metadata 返回 taskId
`server/service/agent/session.ts`	修改	listSessions 过滤 A2A 会话
`server/service/agent/chat-engine.ts`	修改	调用 getAgentToolsForChatByAgentId 时传入 sessionId
`server/api/agents/a2a/stream.get.ts`	新增	A2A SSE 端点
`app/components/agent/AgentInvokePanel.vue`	新增	侧边面板组件
`app/composables/useAgentChat.ts`	修改	检测 agent-invoke tool call，管理面板状态
`app/components/agent/AgentChatArea.vue`	修改	挂载 AgentInvokePanel

4.2 需要确认的依赖

~~executor 的 ctx 是否已包含 sessionId？~~ 已确认不包含，需扩展 ToolContext、executeAgentTool、getAgentToolsForChatByAgentId 接口（见 3.2.4）
AgentInvocationResult 类型需要新增 taskId 字段

5. 开放问题

~~executor ctx.sessionId~~：已解决 — 需扩展 ToolContext、executeAgentTool、getAgentToolsForChatByAgentId 接口。
streamBuffer key 命名：使用 a2a_${taskId} 作为 key，与现有 sessionId 格式 as_xxx 不冲突。
并发 A2A 调用：同一调用方同时发起多个 agent-invoke（多步 tool call），面板如何处理？建议只展示最后一个，或支持多个面板堆叠（后续迭代）。

20 KiB Raw Blame History