在使用 AI Agent 开发复杂任务(如代码重构、长篇写作)时,最头疼的就是“断档”。一旦程序崩溃或需要分阶段执行,AI 往往会丢失之前的上下文。

Anthropic 发布的 Claude Agent SDK 提供了一套原生的 Session(会话)管理机制,支持会话保存、无缝恢复,甚至支持像 Git 分支一样的“会话 Fork”。今天我们就通过 Python 代码来实战演练如何玩转它。


1. 为什么需要 Session 管理?

在默认的 API 调用中,上下文是临时的。而 Agent SDK 的 Session 具有以下核心能力:

  • 状态持久化:获取 session_id,随时中断并在几小时后恢复。
  • 上下文延续:自动加载之前的工具调用结果、思考过程和对话历史。
  • 会话分支 (Forking) :从同一个节点开启两个不同的尝试(例如:方案 A vs 方案 B),互不干扰。

2. 获取并保存 Session ID

当你启动一个 query 时,SDK 会触发一个 init 类型的系统消息。我们需要从中捕获 session_id

Python

import asyncio
from claude_agent_sdk import ClaudeSDKClient

async def start_new_task():
    client = ClaudeSDKClient()
    
    # 开启一个新任务
    response = client.query(
        prompt="帮我设计一个电商系统的数据库表结构",
        model="claude-3-5-sonnet-20241022"
    )

    session_id = None

    async for message in response:
        # 核心:捕获初始化消息中的 session_id
        if message.type == "system" and message.get("subtype") == "init":
            session_id = message.session_id
            print(f" 会话已启动,ID: {session_id}")
            # 你可以将此 ID 存入数据库或本地文件
        
        # 打印 Claude 的思考或回复
        if message.type == "text":
            print(f"Claude: {message.content}")

    return session_id

# 运行
# session_id = asyncio.run(start_new_task())

3. 恢复会话 (Resuming)

有了 session_id,你可以随时“接力”之前的对话。Claude 会记得之前讨论的所有细节。

Python

async def resume_task(old_session_id):
    client = ClaudeSDKClient()
    
    # 使用 resume 参数指定 session_id
    response = client.query(
        prompt="接着上面的设计,请为这些表写出 SQL 创建语句",
        resume=old_session_id,  # 关键参数
        model="claude-3-5-sonnet-20241022"
    )

    async for message in response:
        if message.type == "text":
            print(message.content)

# asyncio.run(resume_task("your-previous-session-id"))

4. 高级操作:会话分支 (Forking)

这是 Agent SDK 最强大的功能之一。假设你让 Claude 写了一个 REST API,现在你想尝试“如果不基于 REST,而是用 GraphQL 怎么做?”,但又不想破坏原来的 REST 会话记录。

你可以通过 fork_session=True 来创建一个镜像分支:

模式fork_session结果
继续 (Continue)False (默认)历史记录追加到原 Session,Session ID 不变。
分支 (Fork)True创建一个新的 Session ID,保留截止到目前的历史,之后的操作不影响原 Session。

代码示例:

Python

async def fork_exploration(original_id):
    client = ClaudeSDKClient()
    
    print("--- 开启 GraphQL 分支 ---")
    # Fork 一个新会话
    forked_response = client.query(
        prompt="现在,尝试把刚才的设计改成 GraphQL 方案",
        resume=original_id,
        fork_session=True, # 开启分支
        model="claude-3-5-sonnet-20241022"
    )

    async for message in forked_response:
        if message.type == "system" and message.get("subtype") == "init":
            print(f" 新的分支 Session ID: {message.session_id}")
        
        if message.type == "text":
            print(message.content)

    # 此时原 original_id 的会话依然停留在 REST 阶段,不受影响

5. 最佳实践提示

  1. 持久化存储:在生产环境中,建议将 session_id 与用户 ID 绑定存储在 Redis 或数据库中。
  2. 工具权限:恢复会话时,记得重新传入 allowed_tools。如果某些敏感工具在恢复时不希望被调用,可以在恢复时通过配置进行限制。
  3. 结合 Checkpointing:Session 管理对话状态,而 File Checkpointing 管理文件变更。两者结合可以实现完整的“实验回滚”机制。

总结

Claude Agent SDK 的 Session 机制让 AI 从“一次性聊天机器人”进化为了“具有长期记忆的工作助手”。通过 resumefork_session,我们可以像管理代码分支一样管理 AI 的思考过程。

你在开发 AI Agent 时遇到过状态丢失的问题吗?欢迎在评论区交流!

本站提供的所有下载资源均来自互联网,仅提供学习交流使用,版权归原作者所有。如需商业使用,请联系原作者获得授权。 如您发现有涉嫌侵权的内容,请联系我们 邮箱:alixiixcom@163.com