이벤트 스트리밍

이벤트 스트리밍 (Event streaming)

Deep Agents에서 서브에이전트, 메시지, 도구 호출, 최종 출력을 스트리밍하세요.

이 페이지는 Deep Agents 특유의 스트리밍 관련 사항을 다룹니다. 가장 중요한 것은 stream.subagents를 통한 위임된 서브에이전트의 스트리밍입니다. 일반 에이전트 스트리밍(stream.messages, stream.values, 도구 호출, 커스텀 업데이트)은 LangChain 이벤트 스트리밍을 참고하세요.

출처: 문서

본문

서브에이전트 스트리밍 (Stream subagents)

Deep Agents는 LangGraph 스트리밍 위에 서브에이전트 프로젝션을 추가합니다. 위임된 각 task 호출에 대해 하나의 스트림 핸들을 원할 때 stream.subagents를 사용하세요. 이 프로젝션은 가볍습니다: 먼저 서브에이전트 작업을 발견하고, 핸들에서 접근할 때만 메시지, 도구 호출, 값 스트림이 열립니다.

각 핸들의 name은 서브에이전트의 구성된 이름입니다: 코디네이터가 task 도구를 호출할 때 전달하는 subagent_type입니다. Deep Agents는 그 이름을 위임된 실행에 바인딩하므로, 서브에이전트 스펙에서 정의한 것과 같은 라벨이 스트림에서 필터링하고 라우팅하는 대상이 됩니다.

const stream = await agent.streamEvents(
  { messages: [{ role: "user", content: "Write me a haiku about the sea" }] },
  { version: "v3" },
);

const subagentNames: string[] = [];
for await (const subagent of stream.subagents) {
  console.log(subagent.name);
  console.log(await subagent.taskInput);

  for await (const message of subagent.messages) {
    console.log(await message.text);
  }

  subagentNames.push(subagent.name);
}

await stream.output;

서브에이전트 스트림 필드 (Subagent stream fields)

각 서브에이전트 스트림은 부모 실행과 같은 종류의 프로젝션(메시지, 도구 호출, 중첩 서브에이전트, 최종 출력)을 노출합니다. 일반 부모 실행 스트리밍 모델은 LangChain 이벤트 스트리밍을 참고하세요.

TypeScript는 toolCalls와 taskInput 같은 camelCase 프로젝션 이름을 사용합니다. 각 서브에이전트 스트림은 .messages, .toolCalls, .values, .subagents, .output을 노출할 수 있습니다.

필드 설명
name 코디네이터가 task 호출에서 선택한 subagent_type에서 가져온 서브에이전트 이름.
messages 서브에이전트가 방출한 메시지.
subagents 중첩 서브에이전트 호출.
output 최종 서브에이전트 상태 또는 위임된 작업의 완료 신호.
taskInput task 도구에 전달된 프롬프트에 대한 Promise.
toolCalls 서브에이전트 범위의 도구 호출.

서브에이전트 수명주기 추적 (Track subagent lifecycle)

어떤 서브에이전트가 시작되고 끝났는지만 보여주면 될 때 stream.subagents를 사용하세요. 개별 서브에이전트에서 해당 프로젝션에 접근하지 않으면 메시지나 값 스트림을 구독할 필요가 없습니다.

const stream = await agent.streamEvents(input, { version: "v3" });

let running = 0;
let completed = 0;
let failed = 0;
const watchers: Promise<void>[] = [];

for await (const subagent of stream.subagents) {
  running += 1;
  console.log(`${subagent.name}: started`);

  watchers.push(
    subagent.output.then(
      () => {
        running -= 1;
        completed += 1;
        console.log(`${subagent.name}: completed`);
      },
      () => {
        running -= 1;
        failed += 1;
        console.log(`${subagent.name}: failed`);
      },
    ),
  );
}

await Promise.all(watchers);
console.log({ running, completed, failed });

메시지 스트리밍 (Stream messages)

Deep Agents는 코디네이터 에이전트와 위임된 서브에이전트에서 메시지를 방출할 수 있습니다. 최상위 메시지에는 stream.messages를, 각 위임된 서브에이전트에는 subagent.messages를 사용하세요.

const stream = await agent.streamEvents(input, { version: "v3" });

const coordinatorMessages: string[] = [];
for await (const message of stream.messages) {
  const text = await message.text;
  console.log("[coordinator]", text);
  coordinatorMessages.push(text);
}

for await (const subagent of stream.subagents) {
  for await (const message of subagent.messages) {
    console.log(`[${subagent.name}]`, await message.text);
  }
}

await stream.output;

도구 호출 스트리밍 (Stream tool calls)

Deep Agents는 에이전트 트리의 각 수준에서 도구 호출을 노출합니다. 코디네이터 도구에는 최상위 stream.tool_calls를, 위임된 작업에는 각 subagent.tool_calls를 사용하세요.

const stream = await agent.streamEvents(input, { version: "v3" });

const coordinatorToolNames: string[] = [];
for await (const call of stream.toolCalls) {
  console.log("[coordinator tool]", call.name, call.input);
  console.log(await call.status);
  coordinatorToolNames.push(call.name);
}

for await (const subagent of stream.subagents) {
  for await (const call of subagent.toolCalls) {
    console.log(`[${subagent.name} tool]`, call.name, call.input);

    const status = await call.status;
    if (status === "finished") {
      console.log(await call.output);
    } else if (status === "error") {
      console.error(await call.error);
    }
  }
}

중첩 작업 스트리밍 (Stream nested work)

서브에이전트 스트림으로 재귀하여 중첩 서브에이전트, 메시지, 도구 호출을 관찰할 수 있습니다.

const stream = await agent.streamEvents(input, { version: "v3" });

const subagentNames: string[] = [];
for await (const subagent of stream.subagents) {
  console.log(`subagent ${subagent.name}: started`);

  for await (const toolCall of subagent.toolCalls) {
    console.log(`${toolCall.name}(${JSON.stringify(toolCall.input)})`);

    const status = await toolCall.status;
    if (status === "finished") {
      console.log(await toolCall.output);
    } else if (status === "error") {
      console.error(await toolCall.error);
    }
  }

  for await (const nested of subagent.subagents) {
    console.log(`nested subagent ${nested.name}: started`);
  }

  subagentNames.push(subagent.name);
}

동시에 소비하기 (Consume concurrently)

코디네이터와 서브에이전트 출력은 종종 서로 섞입니다. 실시간 UI 업데이트가 필요하면 프로젝션을 동시에 소비하세요.

JavaScript에서 동시 소비자를 사용하세요:

const stream = await agent.streamEvents(input, { version: "v3" });

await Promise.all([
  (async () => {
    for await (const message of stream.messages) {
      console.log("[coordinator]", await message.text);
    }
  })(),
  (async () => {
    for await (const subagent of stream.subagents) {
      void (async () => {
        for await (const message of subagent.messages) {
          console.log(`[${subagent.name}]`, await message.text);
        }
      })();
    }
  })(),
]);

코디네이터와 모든 서브에이전트에 걸친 정확한 도착 순서가 필요하면 원시 프로토콜 이벤트를 순회하고 namespace를 사용해 출처를 식별하세요:

const stream = await agent.streamEvents(input, { version: "v3" });

const textDeltas: string[] = [];
for await (const event of stream) {
  if (event.method !== "messages") continue;

  const data = event.params.data;
  if (data.event !== "content-block-delta") continue;

  const block = data.delta ?? {};
  if (block.type === "text-delta") {
    const isSubagent = event.params.namespace.some((seg) =>
      seg.startsWith("tools:"),
    );
    const source = isSubagent ? "subagent" : "coordinator";
    console.log(`[${source}] ${block.text}`);
    textDeltas.push(block.text);
  }
}

서브에이전트 대 서브그래프 (Subagents versus subgraphs)

stream.subgraphs는 그래프 실행 구조를 보여줍니다. stream.subagents는 제품 수준의 Deep Agents 작업 위임을 보여줍니다. 사용자 대면 UI에는 stream.subagents를 사용하세요. 내부 그래프 노드를 숨기고 서브에이전트 개념을 직접 노출하기 때문입니다.

더 알아보기