流式输出

session.stream(prompt) 会在回合运行过程中逐步产出事件，因此你可以在文本到达时即时渲染，并实时响应工具活动。当你需要实时 UI，或希望 CLI 逐 token 打印输出（而不是等待 send 或 run 返回完整结果）时，请使用它。

每个事件都带有稳定的 version、type、payload 和可选 metadata 信封字段。常见类型包括 agent_start、text_delta / reasoning_delta、tool_start / tool_end、agent_end 和 error。验证信息位于 agent_end 的 payload 与便捷字段中。

Rust

Node.js

Python

Rust

use a3s_code_core::{Agent, AgentEvent, CodeError, PlanningMode, SessionOptions};

#[tokio::main]
async fn main() -> a3s_code_core::Result<()> {
    let agent = Agent::new("agent.acl").await?;
    let session = agent
        .session_builder(".")
        .options(SessionOptions::new().with_planning_mode(PlanningMode::Disabled))
        .build()
        .await?;

    let (mut events, lifecycle) = session
        .stream("使用 bash 工具运行测试，然后总结结果。", None)
        .await?;

    while let Some(event) = events.recv().await {
        match event {
            AgentEvent::TextDelta { text } => print!("{text}"),
            AgentEvent::ToolStart { name, .. } => println!("\n[工具开始] {name}"),
            AgentEvent::ToolEnd {
                name, exit_code, ..
            } => println!("\n[工具结束] {name}，退出码={exit_code}"),
            AgentEvent::End {
                verification_summary,
                ..
            } => println!("\n[验证] {verification_summary:?}"),
            AgentEvent::Error { message } => return Err(CodeError::Llm(message)),
            _ => {}
        }
    }
    lifecycle
        .await
        .map_err(|error| CodeError::Internal(error.into()))?;
    println!("\n[流式输出完成]");

    session.close().await;
    agent.close().await;
    Ok(())
}

TypeScript

import { Agent } from '@a3s-lab/code';

const agent = await Agent.create('agent.acl');
const session = await agent.sessionAsync(process.cwd(), {
  planningMode: 'disabled',
});

const stream = await session.stream(
  'Use the bash tool to run the tests, then summarize the result.',
);

for await (const event of stream) {
  if (event.type === 'text_delta' && event.text) {
    process.stdout.write(event.text);
  } else if (event.type === 'tool_start') {
    console.log(`\n[tool:start] ${event.toolName ?? 'unknown'}`);
  } else if (event.type === 'tool_end') {
    console.log(
      `\n[tool:end] ${event.toolName ?? 'unknown'} exit=${event.exitCode ?? 0}`,
    );
  } else if (event.type === 'agent_end') {
    console.log(`\n[verification] ${event.verificationSummaryText ?? ''}`);
  } else if (event.type === 'error') {
    throw new Error(event.error ?? 'stream error');
  }
}

console.log('\n[stream] complete');
await session.closeAsync();
await agent.close();

Python

from a3s_code import Agent, SessionOptions


def main() -> None:
    agent = Agent.create("agent.acl")

    opts = SessionOptions()
    opts.planning_mode = "disabled"
    session = agent.session(".", opts)

    prompt = "Use the bash tool to run the tests, then summarize the result."

    try:
        for event in session.stream(prompt):
            if event.type == "text_delta" and event.text:
                print(event.text, end="", flush=True)
            elif event.type == "tool_start":
                print(f"\n[tool:start] {event.tool_name or 'unknown'}")
            elif event.type == "tool_end":
                print(f"\n[tool:end] {event.tool_name or 'unknown'} exit={event.exit_code or 0}")
            elif event.type == "agent_end":
                print(f"\n[verification] {event.verification_summary_text or ''}")
            elif event.type == "error":
                raise RuntimeError(event.error or "stream error")
        print("\n[stream] complete")
    finally:
        session.close()
        agent.close()


if __name__ == "__main__":
    main()

package main

import (
	"context"
	"fmt"
	"log"

	code "github.com/A3S-Lab/Code/sdk/go/v9"
)

func main() {
	ctx := context.Background()
	agent, err := code.Create(ctx, "agent.acl")
	if err != nil {
		log.Fatal(err)
	}
	defer agent.Close(context.Background())

	session, err := agent.Session(ctx, ".", &code.SessionOptions{
		PlanningMode: code.PlanningDisabled,
	})
	if err != nil {
		log.Fatal(err)
	}
	defer session.Close(context.Background())

	stream, err := session.Stream(
		ctx,
		"使用 bash 工具运行测试，然后总结结果。",
		nil,
	)
	if err != nil {
		log.Fatal(err)
	}
	for event := range stream.Events {
		// 载荷字段与线上事件一致：text、name、exit_code、
		// verification_summary、message。
		var payload struct {
			Text                string                   `json:"text"`
			Name                string                   `json:"name"`
			ExitCode            int                      `json:"exit_code"`
			VerificationSummary code.VerificationSummary `json:"verification_summary"`
			Message             string                   `json:"message"`
		}
		switch event.Type {
		case code.EventTextDelta, code.EventToolStart, code.EventToolEnd,
			code.EventAgentEnd, code.EventError:
			if err := event.DecodePayload(&payload); err != nil {
				log.Fatal(err)
			}
		default:
			// 其他事件（event.Payload、event.Metadata）按需转发。
			continue
		}
		switch event.Type {
		case code.EventTextDelta:
			fmt.Print(payload.Text)
		case code.EventToolStart:
			fmt.Printf("\n[tool:start] %s\n", payload.Name)
		case code.EventToolEnd:
			fmt.Printf("\n[tool:end] %s exit=%d\n", payload.Name, payload.ExitCode)
		case code.EventAgentEnd:
			fmt.Printf("\n[verification] %s\n", payload.VerificationSummary.Status)
		case code.EventError:
			log.Fatal(payload.Message)
		}
	}
	if err := <-stream.Done; err != nil {
		log.Fatal(err)
	}
	fmt.Println("\n[stream] complete")
}

说明：

Rust 从 Tokio channel 接收 AgentEvent，并等待返回的生命周期句柄。Node.js 使用 for await 遍历 EventStream（也可以直接调用 stream.next()）。Python 的 EventStream 同时支持 for 与 async for。Go 必须先持续读取 stream.Events，再从 stream.Done 读取最终错误。取消 Go Context 也会请求原生 Session 取消当前 Run。
四种 SDK 都公开规范事件类型。Node.js 和 Python 提供便捷投影；Go 将 Payload 与 Metadata 保留为 json.RawMessage，并提供 DecodePayload。未来未知事件类型在四种 SDK 中都不会丢失。
tool_start 在模型开始输出工具调用时触发；tool_execution_start 在工具真正开始执行时触发；tool_end 携带工具输出与退出码。
启用确认策略后，流中还会出现人工介入（human-in-the-loop）信号：先是 confirmation_required，收到答复后是 confirmation_received。待确认的调用会一直挂起直到得到答复，不存在计时器替它结算。只有当确认通道在答复前关闭（例如取消或关闭 Session）时，才会发出带 action_taken 的 confirmation_timeout。

可运行的流式示例位于 sdk/node/examples/streaming/。完整的人工确认循环在 sdk/node/examples/streaming/hitl_confirmation_loop.ts，对应 Python 版本位于 sdk/python/examples/，Go 流式行为由 sdk/go/session_test.go 覆盖。

#流式输出

流式输出