Claude Agent SDK 使用指南：流式 vs 单次消息模式 (Streaming vs Single Mode)

在构建 AI Agent 时，如何管理输入和会话状态是核心问题。Anthropic 的 Claude Agent SDK 提供了两种主要的交互模式：流式输入模式 (Streaming Input Mode) 和 单次消息模式 (Single Message Input Mode) 。

本教程将详细解读这两种模式的区别，并提供基于 Python SDK (claude-agent-sdk) 的完整代码示例。

核心概念对比

在 Python SDK 中，这两种模式分别对应了不同的实现方式：

1. 单次消息模式 (Single Message Input)

这是最简单的使用方式。当你只需要 Claude 回答一个问题，或者执行一个独立的任务（如“写一个 Python 脚本”），而不需要它记住之前的对话时，使用 query() 函数。

适用场景

• 一次性的问答
• 不需要上下文记忆的自动化脚本
• Lambda 函数等无状态环境

Python 代码示例

使用 query() 函数，传入提示词即可。注意它返回的是一个异步迭代器。

Python

import asyncio
from claude_agent_sdk import query, ClaudeAgentOptions, AssistantMessage, TextBlock

async def main():
    print("--- 单次模式示例 ---")
    
    # 配置选项（可选）
    options = ClaudeAgentOptions(
        model="claude-3-5-sonnet-20240620",
        max_turns=5
    )

    # 直接调用 query，每次调用都会创建一个全新的会话
    async for message in query(prompt="请用一句话解释什么是量子纠缠", options=options):
        # 过滤并打印 Claude 的文本回复
        if isinstance(message, AssistantMessage):
            for block in message.content:
                if isinstance(block, TextBlock):
                    print(f"Claude: {block.text}")

    # 注意：如果再次调用 query()，Claude 不会记得上面的对话
    print("n--- 新的会话 ---")
    async for message in query(prompt="我刚才问了你什么？"): 
        if isinstance(message, AssistantMessage):
            for block in message.content:
                if isinstance(block, TextBlock):
                    print(f"Claude: {block.text}") 
                    # Claude 会回答不知道，因为这是新会话

if __name__ == "__main__":
    asyncio.run(main())

2. 流式输入模式 (Streaming Input / Persistent Session)

这是开发复杂 Agent 的推荐模式。通过 ClaudeSDKClient，你可以创建一个持久的会话对象。在这个会话中，Claude 可以“记住”之前的交互，支持多轮对话，甚至支持在执行过程中被中断。

适用场景

• 构建聊天机器人 (Chatbot)
• 需要多步推理的复杂任务
• REPL 环境或交互式命令行工具
• 需要根据 Claude 的回复动态调整后续输入的场景

Python 代码示例

使用 ClaudeSDKClient 作为上下文管理器来维护会话。

Python

import asyncio
from claude_agent_sdk import ClaudeSDKClient, AssistantMessage, TextBlock

async def main():
    print("--- 流式/持久会话模式示例 ---")

    # 使用 ClaudeSDKClient 创建一个持久会话
    async with ClaudeSDKClient() as client:
        
        # 第一轮交互
        print("User: 法国的首都是哪里？")
        await client.query("法国的首都是哪里？")
        
        #以此循环接收响应
        async for message in client.receive_response():
            if isinstance(message, AssistantMessage):
                for block in message.content:
                    if isinstance(block, TextBlock):
                        print(f"Claude: {block.text}")

        # 第二轮交互 - Claude 拥有上下文记忆
        print("nUser: 这个城市有多少人口？ (指代巴黎)")
        await client.query("这个城市有多少人口？")
        
        async for message in client.receive_response():
            if isinstance(message, AssistantMessage):
                for block in message.content:
                    if isinstance(block, TextBlock):
                        print(f"Claude: {block.text}")

        # 第三轮交互 - 展示流式输入的强大之处（动态生成消息）
        print("nUser: (发送流式数据...)")
        
        # 定义一个异步生成器作为输入
        async def message_stream():
            yield {"type": "text", "text": "请分析以下数据："}
            await asyncio.sleep(0.5) # 模拟实时数据流
            yield {"type": "text", "text": "温度: 25°C, 湿度: 60%"}
        
        await client.query(message_stream())
        
        async for message in client.receive_response():
            if isinstance(message, AssistantMessage):
                for block in message.content:
                    if isinstance(block, TextBlock):
                        print(f"Claude: {block.text}")

if __name__ == "__main__":
    asyncio.run(main())

进阶技巧：何时选择？

• 选择 query() (Single Mode) ：当你只是想快速测试一个 Prompt，或者你的应用架构本身就是无状态的（REST API 的每个请求都是独立的）。
• 选择 ClaudeSDKClient (Streaming Mode) ：绝大多数 Agent 开发场景。特别是当你需要使用 Tools (工具调用) 时，持久会话能让 Claude 在多次来回交互中修正错误、请求更多信息或分步执行任务。

总结

Claude Agent SDK 通过区分这两种模式，既提供了脚本化的便利性，也保留了构建复杂系统的能力。掌握 ClaudeSDKClient 是构建高级 AI Agent 的第一步。