第 4 章：多轮对话与流式交互

前面几章我们一直在用 query() 函数和 Claude 对话。它简单、好用，但有一个根本性的限制： 每次调用都是独立的，Claude 不记得你之前说了什么。

这一章，我们要解锁一个新能力：让 Claude 在一段持续的对话中记住上下文。

1. query() 的局限：健忘症患者

先来看一个问题。假设你想和 Claude 聊一个话题，先问第一个问题，再基于回答追问：

from claude_agent_sdk import AssistantMessage, TextBlock, query

# 第一次问
async for msg in query(prompt="我叫小明，请记住我的名字"):
    ...

# 第二次问
async for msg in query(prompt="我叫什么名字？"):
    ...
# Claude: "抱歉，我不知道你叫什么名字。"

为什么？因为 query() 的每次调用都是一个全新的进程。就好比你给一个陌生人发短信—— 你发完一条，对方回了，然后你换了个新号码又发了一条。对方当然不知道你是谁。

query() 像发短信 —— 发完就完了，对方不记得你之前说了什么。

ClaudeSDKClient 像打电话 —— 你拨通了，一直保持通话，对方始终记得整个对话的内容。

2. ClaudeSDKClient 入门

最简用法

import asyncio
from claude_agent_sdk import (
    AssistantMessage, ClaudeSDKClient, ResultMessage, TextBlock
)


async def main():
    # async with 自动处理连接和断开
    async with ClaudeSDKClient() as client:
        # 发送消息
        await client.query("你好，我叫小明！")

        # 接收回复
        async for msg in client.receive_response():
            if isinstance(msg, AssistantMessage):
                for block in msg.content:
                    if isinstance(block, TextBlock):
                        print(block.text)


asyncio.run(main())

三个关键步骤：

步骤	代码	说明
1. 建立连接	`async with ClaudeSDKClient() as client`	启动一个持续的会话
2. 发送消息	`await client.query("...")`	把消息发给 Claude
3. 接收回复	`async for msg in client.receive_response()`	逐条接收回复，直到收到 ResultMessage

async with 帮你自动处理了连接（connect()）和断开（disconnect()），就像 Python 的 with open() 自动帮你关文件一样。你不需要操心资源清理的问题。

和 query() 的对比

特性	`query()`	`ClaudeSDKClient`
上下文记忆	无	有
多轮对话	不支持	支持
中断	不支持	支持
动态配置	不支持	支持（切换模型、权限等）
使用难度	简单	稍复杂
适合场景	一问一答	持续交互

3. 多轮对话

ClaudeSDKClient 真正的威力在于：你可以连续发送多个 query()，Claude 会记住之前的所有对话。

import asyncio
from claude_agent_sdk import (
    AssistantMessage, ClaudeSDKClient, ResultMessage, TextBlock
)


def print_response(msg):
    """打印 Claude 回复中的文字内容。"""
    if isinstance(msg, AssistantMessage):
        for block in msg.content:
            if isinstance(block, TextBlock):
                print(f"Claude: {block.text}")


async def main():
    async with ClaudeSDKClient() as client:
        # ---- 第一轮：告诉 Claude 你的名字 ----
        await client.query("我叫小明，我是一个 Python 开发者。请记住这些信息。")
        async for msg in client.receive_response():
            print_response(msg)

        print("---")

        # ---- 第二轮：追问，Claude 应该记得 ----
        await client.query("我叫什么名字？我是做什么的？")
        async for msg in client.receive_response():
            print_response(msg)

        print("---")

        # ---- 第三轮：基于前面的对话继续 ----
        await client.query("给我推荐一个适合我的 Python 项目练手。")
        async for msg in client.receive_response():
            print_response(msg)


asyncio.run(main())

输出大概是这样的：

Claude: 好的，我记住了！你叫小明，是一个 Python 开发者。
---
Claude: 你叫小明，你是一个 Python 开发者。
---
Claude: 基于你是 Python 开发者的背景，我推荐你可以试试...

注意第三轮：Claude 不仅记得你的名字，还记得你是 Python 开发者，所以它的推荐是有针对性的。这就是多轮对话的价值——对话有了连续性，Claude 能基于完整的上下文做出更好的回答。

session_id：给对话起个名字

query() 方法有一个可选参数 session_id，默认值是 "default"。同一个 session_id 下的消息共享上下文：

# 这两条消息在同一个 session 里，Claude 能记住上下文
await client.query("我叫小明", session_id="chat-1")
await client.query("我叫什么？", session_id="chat-1")

# 这条消息在另一个 session 里，Claude 不知道你是谁
await client.query("我叫什么？", session_id="chat-2")

在大多数场景下，你不需要指定 session_id，用默认值就好。它主要用于需要管理多个独立对话的高级场景（比如同时服务多个用户）。

完整代码见 examples/multi_session.py

4. receive_response() vs receive_messages()

ClaudeSDKClient 有两种接收消息的方式，别搞混了：

receive_response() — 推荐，最常用

async for msg in client.receive_response():
    # 逐条处理消息
    # 收到 ResultMessage 后自动停止
    pass

receive_response() 会一直接收消息，直到收到一个 ResultMessage（表示本轮回复结束），然后自动停止迭代。这是你 90% 的场景应该用的。

每次你调用 query() 发一条消息，就用 receive_response() 接收对应的回复。一发一收，节奏清晰。

receive_messages() — 持续监听，不自动停止

async for msg in client.receive_messages():
    # 持续接收所有消息
    # 不会自动停止！需要你自己 break
    if should_stop(msg):
        break

receive_messages() 会无限接收消息，永远不会自动停止。就像开了一个"消息监控器"，所有消息都会经过这里。

什么时候用 receive_messages()？

你想在后台持续监听所有消息
你需要自定义停止条件（比如收到特定内容就停）
你在做更底层的消息处理

日常开发：用 receive_response()。 简单、安全、不会忘记停止。

5. 流式输出 — "打字机"效果

到目前为止，我们处理的都是"完整消息"——Claude 把一整段话想完了，打包发给你。但实际上 Claude 是一个字一个字生成的，你可以利用这个特性做出"打字机"效果，让用户感觉 Claude 在实时打字。

要实现这个效果，你需要开启 部分消息流（partial messages）：

from claude_agent_sdk import ClaudeAgentOptions, ClaudeSDKClient
from claude_agent_sdk.types import StreamEvent

# 关键：设置 include_partial_messages=True
options = ClaudeAgentOptions(
    include_partial_messages=True,
)

开启后，除了完整的 AssistantMessage，你还会收到 StreamEvent 类型的消息。每个 StreamEvent 携带一小块增量数据，对应 Anthropic API 的原始流事件。

从 StreamEvent 中提取文字增量

StreamEvent 的核心结构是 event 字段，它是一个字典，包含 Anthropic API 的原始流事件。最常用的事件类型是 content_block_delta，里面有文字增量：

import asyncio
from claude_agent_sdk import ClaudeAgentOptions, ClaudeSDKClient, ResultMessage
from claude_agent_sdk.types import StreamEvent


async def main():
    options = ClaudeAgentOptions(
        include_partial_messages=True,
    )

    async with ClaudeSDKClient(options) as client:
        await client.query("用三句话介绍 Python 的历史")

        async for msg in client.receive_response():
            if isinstance(msg, StreamEvent):
                event = msg.event
                # 提取文字增量
                if event.get("type") == "content_block_delta":
                    delta = event.get("delta", {})
                    if delta.get("type") == "text_delta":
                        # 逐字打印，不换行
                        print(delta["text"], end="", flush=True)

            elif isinstance(msg, ResultMessage):
                print()  # 最后换行
                print(f"\n(耗时 {msg.duration_ms}ms)")


asyncio.run(main())

运行效果：文字会像打字机一样，一个词一个词地蹦出来，而不是等 Claude 全部想完才一次性显示。

StreamEvent 中的常见事件类型

event.type	说明	包含的数据
`message_start`	消息开始	模型信息、usage 等
`content_block_start`	内容块开始	内容块类型（text/tool_use 等）
`content_block_delta`	内容块增量	delta 中有 text_delta 或 input_json_delta
`content_block_stop`	内容块结束	无额外数据
`message_delta`	消息级别更新	stop_reason、usage 更新
`message_stop`	消息结束	无额外数据

对于打字机效果，你只需要关心 content_block_delta 事件中 delta.type 为 text_delta 的情况（正如上面代码所示）。其他事件类型在做更高级的流处理时才会用到。

完整实现见 examples/typewriter.py

6. 动态配置：运行时切换模型和权限

ClaudeSDKClient 有一个 query() 做不到的能力：在对话过程中动态调整配置。

set_model() — 运行时切换模型

async with ClaudeSDKClient() as client:
    # 先用默认模型（通常是 Sonnet）做快速分析
    await client.query("分析这段代码有什么问题：def foo(): return 1/0")
    async for msg in client.receive_response():
        print_response(msg)

    # 切换到更强的模型做深度优化
    await client.set_model("claude-opus-4-1-20250805")
    await client.query("请给出一个完整的、带错误处理的改进方案")
    async for msg in client.receive_response():
        print_response(msg)

场景：先用便宜模型做粗筛，再用贵的模型做精细活。

比如你在做代码审查：先让 Sonnet 快速扫一遍找出潜在问题，然后对有疑问的地方切换到 Opus 做深入分析。这样既省钱又不牺牲质量。

传 None 可以切回默认模型：

await client.set_model(None)  # 切回默认模型

set_permission_mode() — 运行时切换权限

async with ClaudeSDKClient() as client:
    # 先用安全模式做分析（Claude 不会修改任何文件）
    await client.query("帮我看看这个项目的结构")
    async for msg in client.receive_response():
        print_response(msg)

    # 分析完了，切换到 acceptEdits 模式，允许 Claude 直接修改文件
    await client.set_permission_mode("acceptEdits")
    await client.query("把 README 里的拼写错误修复一下")
    async for msg in client.receive_response():
        print_response(msg)

四种权限模式：

模式	说明
`"default"`	默认模式，遇到危险操作时会暂停等待确认
`"acceptEdits"`	自动接受文件编辑，但其他危险操作仍需确认
`"plan"`	规划模式，Claude 只分析不执行
`"bypassPermissions"`	跳过所有权限检查（谨慎使用！）

安全建议： 先在限制模式下让 Claude 分析，确认方案没问题后再放开权限让它执行。不要一上来就给 bypassPermissions。

7. 中断：按下急刹车

如果 Claude 正在做一个很长的任务，你想让它停下来，可以使用 interrupt()：

import asyncio
from claude_agent_sdk import (
    AssistantMessage, ClaudeSDKClient, ResultMessage, TextBlock
)
from claude_agent_sdk.types import ToolUseBlock


async def main():
    async with ClaudeSDKClient() as client:
        # 给一个会产生很长回复的任务
        await client.query("列出 1 到 1000 的所有质数，每个都要解释为什么是质数")

        turn_count = 0
        async for msg in client.receive_response():
            if isinstance(msg, AssistantMessage):
                turn_count += 1

                # 汇总这一轮在干什么
                for block in msg.content:
                    if isinstance(block, TextBlock):
                        print(f"[轮 {turn_count} - 文字] {block.text[:100]}")
                    elif isinstance(block, ToolUseBlock):
                        print(f"[轮 {turn_count} - 工具] 调用 {block.name}")
                    else:
                        # ThinkingBlock 等其他类型
                        print(f"[轮 {turn_count} - {type(block).__name__}]")

                if turn_count >= 3:
                    print("\n--- 够了，打断它！---")
                    await client.interrupt()

            elif isinstance(msg, ResultMessage):
                print(f"\n任务结束，共 {msg.num_turns} 轮")


asyncio.run(main())

每轮 AssistantMessage 的 content 中可能包含不同类型的内容块： - TextBlock — Claude 输出的文字 - ToolUseBlock — Claude 调用工具（如 Bash），可以通过 block.name 看到工具名

这里对每一轮都计数，并打印该轮的具体行为（文字还是工具调用），这样你能清楚看到 Claude 的完整工作过程。

interrupt() 必须在你同时正在消费消息（receive_response() 或 receive_messages()）的时候调用才有效。 因为中断信号需要通过消息通道来处理。

在上面的例子里，我们在 async for 循环内部调用 interrupt()，这是正确的做法。如果你在循环外面调用，中断信号可能无法被正确处理。

更实际的中断场景：超时

import asyncio
from claude_agent_sdk import ClaudeSDKClient, ResultMessage


async def query_with_timeout(client, prompt, timeout_seconds=30):
    """带超时的查询。"""
    await client.query(prompt)

    async def _watchdog():
        """超时后自动中断。"""
        await asyncio.sleep(timeout_seconds)
        await client.interrupt()

    # 启动超时看门狗
    watchdog_task = asyncio.create_task(_watchdog())

    try:
        async for msg in client.receive_response():
            # 正常处理消息...
            yield msg
    finally:
        watchdog_task.cancel()

8. 手动连接管理

async with 很方便，但有时你需要更细粒度的控制——比如在一个长时间运行的程序里，连接的建立和断开发生在不同的函数中。

这时候你可以手动调用 connect() 和 disconnect()：

import asyncio
from claude_agent_sdk import (
    AssistantMessage, ClaudeSDKClient, ResultMessage, TextBlock
)


async def main():
    client = ClaudeSDKClient()

    try:
        # 手动连接
        await client.connect()

        # 发送和接收
        await client.query("你好！")
        async for msg in client.receive_response():
            if isinstance(msg, AssistantMessage):
                for block in msg.content:
                    if isinstance(block, TextBlock):
                        print(block.text)

    finally:
        # 无论是否出错，确保断开连接
        await client.disconnect()


asyncio.run(main())

关键：disconnect() 必须放在 finally 里。 否则如果中间出了异常，连接就泄漏了。这就是为什么大多数时候推荐用 async with——它帮你保证了这一点。

connect() 的高级用法

connect() 可以接受一个初始 prompt，在连接的同时就开始对话：

# 方式一：先连接，再发消息（推荐）
await client.connect()
await client.query("你好")

# 方式二：连接时就发消息
await client.connect(prompt="你好")

两者效果一样。方式一更灵活，方式二更简洁。

不传参数调用 connect() 时，SDK 会以"空流"模式连接——建立好通道，但不发送任何消息，等你后续调用 query() 再开始对话。这也是 async with 内部的默认行为。

9. 完整示例：聊天 REPL

把这一章学到的东西组合起来，做一个简单的聊天程序—— 你在终端输入消息，Claude 实时回复，就像在用 ChatGPT 一样：

import asyncio
from claude_agent_sdk import (
    AssistantMessage, ClaudeSDKClient, ResultMessage, TextBlock
)


async def main():
    print("MiniChat - 和 Claude 聊天（输入 /quit 退出）\n")

    async with ClaudeSDKClient() as client:
        while True:
            user_input = input("你: ").strip()
            if user_input in ("/quit", "/exit", ""):
                print("再见！")
                break

            await client.query(user_input)

            print("Claude: ", end="", flush=True)
            async for msg in client.receive_response():
                if isinstance(msg, AssistantMessage):
                    for block in msg.content:
                        if isinstance(block, TextBlock):
                            print(block.text, end="", flush=True)
                elif isinstance(msg, ResultMessage):
                    print()  # 换行
            print()


asyncio.run(main())

试试运行它，体验多轮对话的效果：

MiniChat - 和 Claude 聊天（输入 /quit 退出）

你: 我叫小明，记住我的名字
Claude: 好的，小明！我记住了。有什么我能帮你的吗？

你: 我叫什么？
Claude: 你叫小明呀！

你: /quit
再见！

完整代码见 examples/chat_loop.py

本章完整示例代码

以下是本章涉及的完整示例文件，可以直接复制运行。

chat_loop.py — 多轮对话 REPL

import asyncio

from claude_agent_sdk import (
    AssistantMessage,
    ClaudeSDKClient,
    ResultMessage,
    TextBlock,
)


async def main():
    print("MiniChat - 和 Claude 聊天（输入 /quit 退出）\n")

    # async with 自动处理连接和断开
    async with ClaudeSDKClient() as client:
        while True:
            # 读取用户输入
            user_input = input("你: ").strip()
            if user_input in ("/quit", "/exit", ""):
                print("再见！")
                break

            # 发送消息给 Claude
            await client.query(user_input)

            # 接收并打印回复
            print("Claude: ", end="", flush=True)
            async for msg in client.receive_response():
                if isinstance(msg, AssistantMessage):
                    # 提取文字内容
                    for block in msg.content:
                        if isinstance(block, TextBlock):
                            print(block.text, end="", flush=True)
                elif isinstance(msg, ResultMessage):
                    # ResultMessage 表示本轮回复结束
                    print()
            print()  # 空行，方便阅读


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

运行方式：

python examples/chat_loop.py

multi_session.py — 多会话与 session_id

import asyncio

from claude_agent_sdk import (
    AssistantMessage,
    ClaudeAgentOptions,
    ClaudeSDKClient,
    ResultMessage,
    TextBlock,
)


async def print_reply(client: ClaudeSDKClient, label: str = "Claude") -> None:
    """接收并打印一轮回复。"""
    print(f"{label}: ", end="", flush=True)
    async for msg in client.receive_response():
        if isinstance(msg, AssistantMessage):
            for block in msg.content:
                if isinstance(block, TextBlock):
                    print(block.text, end="", flush=True)
        elif isinstance(msg, ResultMessage):
            print()
            print(f"  (耗时 {msg.duration_ms}ms, "
                  f"花费 ${msg.total_cost_usd or 0:.4f})")
    print()


async def demo_multi_turn():
    """演示多轮对话的上下文保持。"""
    print("=" * 50)
    print("演示 1：多轮对话 —— Claude 记住上下文")
    print("=" * 50)
    print()

    # 限制轮数防止 Claude 调用工具跑飞
    options = ClaudeAgentOptions(max_turns=1)

    async with ClaudeSDKClient(options) as client:
        # 第一轮：告诉 Claude 一些信息
        print("你: 请记住这三个数字：42, 73, 100。")
        await client.query("请记住这三个数字：42, 73, 100。")
        await print_reply(client)

        # 第二轮：追问——Claude 应该能回忆起来
        print("你: 刚才我让你记住的三个数字是什么？它们的和是多少？")
        await client.query("刚才我让你记住的三个数字是什么？它们的和是多少？")
        await print_reply(client)

        # 第三轮：继续追问——Claude 仍然有完整的上下文
        print("你: 这三个数字中最大的是哪个？")
        await client.query("这三个数字中最大的是哪个？")
        await print_reply(client)


async def demo_session_id():
    """演示 session_id 的隔离效果。"""
    print()
    print("=" * 50)
    print("演示 2：session_id —— 不同会话互相隔离")
    print("=" * 50)
    print()

    options = ClaudeAgentOptions(max_turns=1)

    async with ClaudeSDKClient(options) as client:
        # 在 session "alice" 中设置信息
        print('[session=alice] 你: 我叫 Alice，我喜欢 Python。')
        await client.query("我叫 Alice，我喜欢 Python。", session_id="alice")
        await print_reply(client, label="[session=alice] Claude")

        # 在 session "bob" 中设置不同的信息
        print('[session=bob] 你: 我叫 Bob，我喜欢 Rust。')
        await client.query("我叫 Bob，我喜欢 Rust。", session_id="bob")
        await print_reply(client, label="[session=bob] Claude")

        # 在 session "alice" 中追问——应该知道 Alice 的信息
        print('[session=alice] 你: 我叫什么名字？我喜欢什么编程语言？')
        await client.query(
            "我叫什么名字？我喜欢什么编程语言？", session_id="alice"
        )
        await print_reply(client, label="[session=alice] Claude")

        # 在 session "bob" 中追问——应该知道 Bob 的信息
        print('[session=bob] 你: 我叫什么名字？我喜欢什么编程语言？')
        await client.query(
            "我叫什么名字？我喜欢什么编程语言？", session_id="bob"
        )
        await print_reply(client, label="[session=bob] Claude")


async def main():
    await demo_multi_turn()
    await demo_session_id()


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

运行方式：

python examples/multi_session.py

typewriter.py — 打字机效果（流式输出）

import asyncio
import sys

from claude_agent_sdk import (
    AssistantMessage,
    ClaudeAgentOptions,
    ClaudeSDKClient,
    ResultMessage,
    TextBlock,
)
from claude_agent_sdk.types import StreamEvent


def handle_stream_event(event: dict) -> None:
    """处理单个流事件，提取并打印文字增量。

    Anthropic API 的流事件类型很多，这里只关心文字增量。
    其他事件类型（如 content_block_start、message_delta 等）
    在需要更精细控制时才会用到。
    """
    event_type = event.get("type")

    if event_type == "content_block_delta":
        delta = event.get("delta", {})

        # 文字增量：逐字打印
        if delta.get("type") == "text_delta":
            text = delta.get("text", "")
            sys.stdout.write(text)
            sys.stdout.flush()

        # 思考增量：如果开启了 thinking，也会有这种类型
        elif delta.get("type") == "thinking_delta":
            # 可以选择显示或忽略思考过程
            thinking = delta.get("thinking", "")
            sys.stdout.write(f"\033[90m{thinking}\033[0m")  # 灰色显示思考
            sys.stdout.flush()


async def typewriter_chat():
    """打字机效果的聊天演示。"""
    print("Typewriter Demo - 打字机效果演示")
    print("输入消息和 Claude 聊天，感受实时流式输出")
    print("输入 /quit 退出\n")

    # 关键配置：开启部分消息流
    options = ClaudeAgentOptions(
        include_partial_messages=True,
    )

    async with ClaudeSDKClient(options) as client:
        while True:
            user_input = input("你: ").strip()
            if user_input in ("/quit", "/exit", ""):
                print("再见！")
                break

            await client.query(user_input)

            print("Claude: ", end="", flush=True)

            # 统计信息
            char_count = 0

            async for msg in client.receive_response():
                if isinstance(msg, StreamEvent):
                    # 流事件：逐字输出（打字机效果的核心）
                    handle_stream_event(msg.event)

                elif isinstance(msg, AssistantMessage):
                    # 完整消息：在流式模式下，这是最终的完整消息
                    # StreamEvent 已经把内容逐字打印了，所以这里不需要再打印
                    # 但可以用它来做一些统计
                    for block in msg.content:
                        if isinstance(block, TextBlock):
                            char_count += len(block.text)

                elif isinstance(msg, ResultMessage):
                    # 本轮回复结束
                    print()  # 换行
                    print(
                        f"  [{char_count} 字 | "
                        f"{msg.duration_ms}ms | "
                        f"${msg.total_cost_usd or 0:.4f}]"
                    )

            print()


async def simple_typewriter():
    """最简打字机效果：只用 10 行核心代码。"""
    print("\n--- 最简打字机效果 ---\n")

    options = ClaudeAgentOptions(include_partial_messages=True)

    async with ClaudeSDKClient(options) as client:
        await client.query("用两句话介绍 Python 语言的特点")

        print("Claude: ", end="", flush=True)
        async for msg in client.receive_response():
            if isinstance(msg, StreamEvent):
                event = msg.event
                if event.get("type") == "content_block_delta":
                    delta = event.get("delta", {})
                    if delta.get("type") == "text_delta":
                        print(delta["text"], end="", flush=True)
            elif isinstance(msg, ResultMessage):
                print("\n")


async def main():
    # 先跑一个简单的演示
    await simple_typewriter()

    # 再进入交互式聊天
    await typewriter_chat()


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

运行方式：

python examples/typewriter.py

10. 小结

这一章我们从 query() 的"一次性对话"升级到了 ClaudeSDKClient 的"持续会话"。核心知识点：

ClaudeSDKClient 保持对话上下文 — 多次 query() 之间，Claude 记得之前说了什么
receive_response() 是首选 — 接收一轮完整回复后自动停止，安全省心
receive_messages() 用于持续监听 — 不自动停止，需要你自己控制退出
流式输出用 StreamEvent — 开启 include_partial_messages=True，逐字接收
动态配置 — set_model() 和 set_permission_mode() 随时调整
interrupt() 打断任务 — 必须在消费消息的同时调用
手动管理用 try/finally — 不用 async with 时确保 disconnect()

下一章，我们将学习如何给 Claude 添加自定义工具——让它不仅能说，还能做。

本章文件清单

04-多轮对话与流式交互/
  README.md                          # 你正在读的这个文件
  examples/
    chat_loop.py                     # 聊天 REPL：多轮对话循环
    multi_session.py                 # session_id 用法与多轮上下文
    typewriter.py                    # 打字机效果：流式输出实现