mcpField
L
Letta Conversations API

Letta Conversations API

github.com/letta-aiAgent 工作流152161 次安装4 周前更新
代码仓库

关于 Letta Conversations API

Letta Conversations API 技能介绍如何为单个 Agent 创建相互隔离的对话线程,同时共享记忆块和工具。适用于多用户聊天、会话管理及对话测试。

主要功能

  1. 01隔离各对话的消息历史
  2. 02跨对话共享记忆块和工具
  3. 03流式发送消息
  4. 04分页列出消息和对话
  5. 05提供交互式 CLI 示例

适用场景

  1. 01为多用户聊天应用分配独立线程
  2. 02管理彼此隔离的会话上下文
  3. 03在独立对话中比较 Agent 回复

安装

复制下面的命令,在终端中运行。

$npx skills add https://github.com/letta-ai/skills --skill letta-conversations-api

安装后在 Claude Code 中输入/letta-conversations-api确认已加载。

包含的文件

除 SKILL.md 外,安装时一起下载的参考资料和脚本。

LICENSE
scripts/
conversations_cli.py
conversations_demo.py

详细介绍

Letta 对话 API

对话 API 支持在单个 Agent 上创建多个彼此隔离的消息线程。每个对话都有独立的消息历史,同时共享 Agent 的记忆块和工具。

何时使用此技能

  • 构建多用户聊天应用(每位用户拥有自己的对话)
  • 实现具有独立上下文的会话管理
  • 在彼此隔离的对话中对 Agent 的响应进行 A/B 测试
  • 任何需要让一个 Agent 支持多个独立聊天线程的场景

核心概念

概念描述
对话Agent 上彼此隔离的消息线程(conv-xxx ID)
隔离每个对话都有独立的消息历史
共享状态各对话共享记忆块和工具
上下文内消息当前位于对话上下文窗口中的消息

Python SDK 用法

设置
from letta_client import Letta

client = Letta(base_url="https://api.letta.com", api_key="your-key")
创建对话
conversation = client.conversations.create(agent_id="agent-xxx")
# conversation.id -> "conv-xxx"
发送消息(流式传输)
stream = client.conversations.messages.create(
    conversation_id=conversation.id,
    messages=[{"role": "user", "content": "Hello!"}],
)

for msg in stream:
    if hasattr(msg, "message_type") and msg.message_type == "assistant_message":
        print(msg.content)
列出对话中的消息
messages = client.conversations.messages.list(
    conversation_id=conversation.id,
    limit=50,  # Optional: default 100
    after="message-xxx",  # Optional: cursor for pagination
    before="message-yyy",  # Optional: cursor for pagination
)
列出 Agent 的所有对话
conversations = client.conversations.list(
    agent_id="agent-xxx",
    limit=50,  # Optional
    after="conv-xxx",  # Optional: cursor for pagination
)
获取特定对话
conv = client.conversations.retrieve(conversation_id="conv-xxx")
# conv.in_context_message_ids -> list of message IDs in context window

REST API 端点

方法端点描述
POST/v1/conversations?agent_id=xxx创建对话
GET/v1/conversations?agent_id=xxx列出对话
GET/v1/conversations/{conversation_id}获取对话
GET/v1/conversations/{conversation_id}/messages列出消息
POST/v1/conversations/{conversation_id}/messages发送消息(流式传输响应)
POST/v1/conversations/{conversation_id}/stream恢复后台流
REST 示例:创建对话并发送消息
# Create conversation
curl -X POST "https://api.letta.com/v1/conversations?agent_id=agent-xxx" \
  -H "Authorization: Bearer $LETTA_API_KEY" \
  -H "Content-Type: application/json"

# Send message (streaming response)
curl -X POST "https://api.letta.com/v1/conversations/conv-xxx/messages" \
  -H "Authorization: Bearer $LETTA_API_KEY" \
  -H "Content-Type: application/json" \
  -H "Accept: text/event-stream" \
  -d '{"messages": [{"role": "user", "content": "Hello!"}]}'

对话架构

class Conversation:
    id: str                      # "conv-xxx"
    agent_id: str                # Associated agent ID
    created_at: datetime         # Creation timestamp
    summary: Optional[str]       # Optional conversation summary
    in_context_message_ids: List[str]  # Message IDs in context window

常见模式

多用户聊天应用
# Each user gets their own conversation
user_conversations = {}

def get_or_create_conversation(user_id: str, agent_id: str) -> str:
    if user_id not in user_conversations:
        conv = client.conversations.create(agent_id=agent_id)
        user_conversations[user_id] = conv.id
    return user_conversations[user_id]

def send_user_message(user_id: str, agent_id: str, message: str):
    conv_id = get_or_create_conversation(user_id, agent_id)
    return client.conversations.messages.create(
        conversation_id=conv_id,
        messages=[{"role": "user", "content": message}],
    )
分页浏览消息历史
def get_all_messages(conversation_id: str):
    all_messages = []
    after = None
    
    while True:
        batch = client.conversations.messages.list(
            conversation_id=conversation_id,
            limit=100,
            after=after,
        )
        if not batch:
            break
        all_messages.extend(batch)
        after = batch[-1].id
    
    return all_messages

重要说明

  1. 默认采用流式传输:messages.create 端点始终以流式方式传输响应
  2. 共享记忆:在一个对话中更新的记忆块,在该 Agent 的所有对话中均可见
  3. 消息隔离:各对话的消息历史完全隔离
  4. 分页:使用 after/before 游标进行高效分页,不要使用偏移量

示例脚本

此技能在 scripts/ 目录中包含两个示例脚本:

  1. conversations_demo.py - 全面演示所有 API 功能

    • 基本对话流程
    • 对话隔离测试
    • 列出和获取对话
    • 分页示例
    • 共享记忆演示
  2. conversations_cli.py - 用于管理对话的交互式 TUI

    • 创建对话并在对话之间切换
    • 发送消息并流式接收响应
    • 查看消息历史
    • 在 Agent 之间切换
运行示例
# Run the demo script
LETTA_API_KEY=your-key uv run letta/conversations/scripts/conversations_demo.py

# Run the interactive CLI
LETTA_API_KEY=your-key uv run letta/conversations/scripts/conversations_cli.py

# CLI with specific agent
LETTA_API_KEY=your-key uv run letta/conversations/scripts/conversations_cli.py --agent agent-xxx

SDK 使用注意事项

  • 分页响应使用 .items 访问列表:client.agents.list().items
  • 身份验证参数是 api_key,不是 token:Letta(base_url=..., api_key=...)
  • 必须读取完消息流(通过迭代或调用 list()),请求才能完成

评论

支持 Markdown