流式输出

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 = agent.session(process.cwd(), { planningMode: 'disabled' });

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

while (true) {
  const next = await stream.next();
  if (next.done || !next.value) break;

  const event = next.value;
  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');
session.close();

Python

import os

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()


if __name__ == "__main__":
    main()

package main

import (
	"context"
	"fmt"
	"log"

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

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 {
		switch event.Type {
		case code.EventTextDelta:
			var payload struct {
				Text string `json:"text"`
			}
			if err := event.DecodePayload(&payload); err != nil {
				log.Fatal(err)
			}
			fmt.Print(payload.Text)
		default:
			// 按需把 event.Payload 与 event.Metadata 转发给界面。
		}
	}
	if err := <-stream.Done; err != nil {
		log.Fatal(err)
	}
}

说明：

Rust 从 Tokio channel 接收 AgentEvent；Node.js 使用 stream.next()，Python 使用同步迭代器；Go 必须先持续读取 stream.Events，再从 stream.Done 读取最终错误。取消 Go Context 也会请求原生 Session 取消当前 Run。
四种 SDK 都公开规范事件类型。Node.js 和 Python 提供便捷投影；Go 将 Payload 与 Metadata 保留为 json.RawMessage，并提供 DecodePayload。未来未知事件类型在四种 SDK 中都不会丢失。
当启用确认策略时，流式事件还可能包含人工介入（human-in-the-loop）确认信号（confirmation_required、confirmation_received、confirmation_timeout）。

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

#流式输出

流式输出