Streaming Input

ทำความเข้าใจสอง input modes ของ Claude Agent SDK และเวลาที่ควรใช้แต่ละอัน

ภาพรวม

Claude Agent SDK รองรับสอง input modes ที่แตกต่างกันสำหรับการโต้ตอบกับ agents:

• Streaming Input Mode (ค่าเริ่มต้นและแนะนำ) - เซสชันแบบ persistent และ interactive
• Single Message Input - One-shot queries ที่ใช้ session state และการ resume

คู่มือนี้อธิบายความแตกต่าง, ประโยชน์, และ use cases สำหรับแต่ละ mode เพื่อช่วยให้คุณเลือกแนวทางที่เหมาะสมสำหรับแอปพลิเคชันของคุณ

Streaming Input Mode (แนะนำ)

Streaming input mode เป็นวิธี ที่แนะนำ ในการใช้ Claude Agent SDK ให้การเข้าถึง capabilities ของ agent อย่างเต็มรูปแบบและเปิดใช้งานประสบการณ์ที่ rich และ interactive

ช่วยให้ agent ทำงานเป็น process ที่ทำงานนานซึ่งรับ user input, จัดการการหยุดชะงัก, แสดง permission requests, และจัดการ session management

วิธีที่มันทำงาน

ประโยชน์

• Image Uploads: แนบ images โดยตรงกับ messages สำหรับการวิเคราะห์ภาพ
• Queued Messages: ส่ง messages หลายรายการที่ประมวลผลตามลำดับ พร้อมความสามารถในการหยุดชะงัก
• Tool Integration: เข้าถึง tools ทั้งหมดและ custom MCP servers ในระหว่างเซสชัน
• Real-time Feedback: ดู responses ขณะที่กำลังถูกสร้าง ไม่ใช่แค่ผลลัพธ์สุดท้าย
• Context Persistence: รักษา conversation context ในหลาย turns อย่างเป็นธรรมชาติ

ตัวอย่างการ Implement

import { query, type SDKUserMessage } from "@anthropic-ai/claude-agent-sdk";
import { readFile } from "fs/promises";

async function* generateMessages(): AsyncGenerator<SDKUserMessage> {
  // First message
  yield {
    type: "user",
    message: {
      role: "user",
      content: "Analyze this codebase for security issues"
    },
    parent_tool_use_id: null
  };

  // Wait for conditions or user input
  await new Promise((resolve) => setTimeout(resolve, 2000));

  // Follow-up with image
  yield {
    type: "user",
    message: {
      role: "user",
      content: [
        {
          type: "text",
          text: "Review this architecture diagram"
        },
        {
          type: "image",
          source: {
            type: "base64",
            media_type: "image/png",
            data: await readFile("diagram.png", "base64")
          }
        }
      ]
    },
    parent_tool_use_id: null
  };
}

// Process streaming responses
for await (const message of query({
  prompt: generateMessages(),
  options: {
    maxTurns: 10,
    allowedTools: ["Read", "Grep"]
  }
})) {
  if (message.type === "result" && message.subtype === "success") {
    console.log(message.result);
  }
}

from claude_agent_sdk import (
    ClaudeSDKClient,
    ClaudeAgentOptions,
    AssistantMessage,
    TextBlock,
)
import asyncio
import base64


async def streaming_analysis():
    async def message_generator():
        # First message
        yield {
            "type": "user",
            "message": {
                "role": "user",
                "content": "Analyze this codebase for security issues",
            },
        }

        # Wait for conditions
        await asyncio.sleep(2)

        # Follow-up with image
        with open("diagram.png", "rb") as f:
            image_data = base64.b64encode(f.read()).decode()

        yield {
            "type": "user",
            "message": {
                "role": "user",
                "content": [
                    {"type": "text", "text": "Review this architecture diagram"},
                    {
                        "type": "image",
                        "source": {
                            "type": "base64",
                            "media_type": "image/png",
                            "data": image_data,
                        },
                    },
                ],
            },
        }

    # Use ClaudeSDKClient for streaming input
    options = ClaudeAgentOptions(max_turns=10, allowed_tools=["Read", "Grep"])

    async with ClaudeSDKClient(options) as client:
        # Send streaming input
        await client.query(message_generator())

        # Process responses
        async for message in client.receive_response():
            if isinstance(message, AssistantMessage):
                for block in message.content:
                    if isinstance(block, TextBlock):
                        print(block.text)


asyncio.run(streaming_analysis())

note

ใน TypeScript SDK หาก message generator ของคุณ throw เช่น เมื่อไฟล์ที่อ่านหายไป stream จะสิ้นสุดด้วย error ที่ระบุว่า Claude Code process aborted by user แทนที่จะเป็น error ต้นฉบับ ดังนั้นตรวจสอบโค้ดภายใน generator ของคุณก่อนเมื่อเห็นข้อความนั้น

ใน Python SDK ข้อยกเว้นจาก generator จะถูก log ที่ระดับ debug และเซสชันจะหยุดชะงักโดยไม่ raise ดังนั้นหากเซสชัน streaming หยุดอยู่โดยไม่มี output ให้เปิดใช้งาน debug logging และตรวจสอบ generator ของคุณ

Single Message Input

Single message input เรียบง่ายกว่าแต่มีข้อจำกัดมากกว่า

เมื่อใดควรใช้ Single Message Input

ใช้ single message input เมื่อ:

• คุณต้องการ one-shot response
• คุณไม่ต้องการ image attachments หรือวิธีการควบคุมกลางเซสชัน
• คุณต้องการทำงานในสภาพแวดล้อมแบบ stateless เช่น lambda function

ข้อจำกัด

warning

Single message input mode ไม่รองรับ:

• การแนบ images โดยตรงในข้อความ
• Dynamic message queueing
• การหยุดชะงักแบบ real-time
• การสนทนา multi-turn แบบธรรมชาติ

หาก query จบด้วย error result เช่น error_max_turns การเรียก query() แบบ single message จะ raise error ที่มีข้อความความล้มเหลวหลังจาก yield result message สุดท้าย ดังนั้น wrap loop ใน try block หากโค้ดของคุณต้องการ continue

ตัวอย่างการ Implement

import { query } from "@anthropic-ai/claude-agent-sdk";

// Simple one-shot query
for await (const message of query({
  prompt: "Explain the authentication flow",
  options: {
    maxTurns: 1,
    allowedTools: ["Read", "Grep"]
  }
})) {
  if (message.type === "result" && message.subtype === "success") {
    console.log(message.result);
  }
}

// Continue conversation with session management
for await (const message of query({
  prompt: "Now explain the authorization process",
  options: {
    continue: true,
    maxTurns: 1
  }
})) {
  if (message.type === "result" && message.subtype === "success") {
    console.log(message.result);
  }
}

from claude_agent_sdk import query, ClaudeAgentOptions, ResultMessage
import asyncio


async def single_message_example():
    # Simple one-shot query using query() function
    async for message in query(
        prompt="Explain the authentication flow",
        options=ClaudeAgentOptions(max_turns=1, allowed_tools=["Read", "Grep"]),
    ):
        if isinstance(message, ResultMessage):
            print(message.result)

    # Continue conversation with session management
    async for message in query(
        prompt="Now explain the authorization process",
        options=ClaudeAgentOptions(continue_conversation=True, max_turns=1),
    ):
        if isinstance(message, ResultMessage):
            print(message.result)


asyncio.run(single_message_example())