概览

本指南带你构建一个使用 Synapse 跨会话持久化上下文的 LLM Agent。完成时，你的 Agent 将能够：

• 会话开始时回放过往上下文
• 随时存储新学到的内容
• 跨会话跟踪多步任务
• 通过异步聊天与人类通信

架构

┌──────────────┐   回放/存储    ┌──────────┐
│  LLM Agent   │ ◀──────────────▶ │ Synapse  │
│ (你的代码)   │                  │   API    │
└──────────────┘                  └──────────┘
       │
       │ 轮询/回复
       ▼
┌──────────────┐
│    人类      │ (浏览器或聊天 UI)
└──────────────┘

第 1 步：设置 Mind Key

# 注册并获取 JWT
JWT=$(curl -s -X POST https://synapse.schaefer.zone/register \
  -H "Content-Type: application/json" \
  -d '{"email":"agent@example.com","password":"secret"}' | jq -r .jwt)

# 创建 Mind 并获取 Mind Key
MIND_KEY=$(curl -s -X POST https://synapse.schaefer.zone/minds \
  -H "Authorization: Bearer $JWT" \
  -H "Content-Type: application/json" \
  -d '{"name":"persistent-agent","description":"My persistent agent"}' | jq -r .mind_key)

echo "Save this: $MIND_KEY"

第 2 步：会话启动协议

每次会话开始时，回放所有记忆：

import os
import requests

MIND_KEY = os.environ["SYNAPSE_MIND_KEY"]
URL = "https://synapse.schaefer.zone"

def session_start():
    """每次会话开始时调用。"""
    # 1. 回放所有记忆
    r = requests.get(
        f"{URL}/memory/recall",
        headers={"Authorization": f"Bearer {MIND_KEY}"}
    )
    memories = r.text  # 纯文本摘要
    
    # 2. 检查未读聊天消息
    r = requests.get(
        f"{URL}/chat/poll",
        headers={"Authorization": f"Bearer {MIND_KEY}"}
    )
    messages = r.json().get("messages", [])
    
    # 3. 检查进行中的任务
    r = requests.get(
        f"{URL}/mind/tasks?status=in_progress",
        headers={"Authorization": f"Bearer {MIND_KEY}"}
    )
    tasks = r.json().get("tasks", [])
    
    return {
        "memories": memories,
        "unread_messages": messages,
        "active_tasks": tasks,
    }

context = session_start()
# 用此上下文构建系统 prompt

第 3 步：存储新学习内容

每当 Agent 学到值得记住的内容：

def remember(category, key, content, tags=None, priority="normal"):
    """存储一条记忆。"""
    requests.post(
        f"{URL}/memory",
        headers={
            "Authorization": f"Bearer {MIND_KEY}",
            "Content-Type": "application/json",
        },
        json={
            "category": category,
            "key": key,
            "content": content,
            "tags": tags or [],
            "priority": priority,
        }
    )

# 示例
remember("identity", "user_name", "User is Michael Schäfer", 
         tags=["person"], priority="critical")
remember("preference", "communication_style", 
         "User prefers concise technical responses",
         tags=["communication"])
remember("project", "current_project", 
         "Building Synapse v1.6.0 with docs system",
         tags=["synapse", "docs"], priority="high")
remember("mistake", "npm_version_bump", 
         "Always bump package.json version after changes",
         tags=["npm", "ci"], priority="high")

第 4 步：任务管理

跨会话跟踪多步工作：

def create_task(title, description="", priority="normal"):
    r = requests.post(
        f"{URL}/mind/task",
        headers={"Authorization": f"Bearer {MIND_KEY}",
                 "Content-Type": "application/json"},
        json={"title": title, "description": description, "priority": priority}
    )
    return r.json()["id"]

def update_task(task_id, status=None, description=None):
    payload = {}
    if status: payload["status"] = status
    if description: payload["description"] = description
    requests.put(
        f"{URL}/mind/task/{task_id}",
        headers={"Authorization": f"Bearer {MIND_KEY}",
                 "Content-Type": "application/json"},
        json=payload
    )

# 跨多会话工作流
task_id = create_task("Deploy v1.6.0", "Push docs system to production", "high")
update_task(task_id, status="in_progress")
# ... 跨多个会话工作 ...
update_task(task_id, status="done")

第 5 步：与人类异步聊天

在工具调用之间轮询消息：

import time

def poll_messages():
    r = requests.get(
        f"{URL}/chat/poll",
        headers={"Authorization": f"Bearer {MIND_KEY}"}
    )
    return r.json().get("messages", [])

def reply(content):
    requests.post(
        f"{URL}/chat/reply",
        headers={"Authorization": f"Bearer {MIND_KEY}",
                 "Content-Type": "application/json"},
        json={"content": content}
    )

# 主循环
while working:
    # 轮询人类消息
    for msg in poll_messages():
        print(f"Human: {msg['content']}")
        reply(f"Got it: {msg['content']}. Working on it.")
    
    # 执行一个工作单元
    do_work()
    
    time.sleep(30)  # 不要轮询过于频繁

第 6 步：会话结束协议

会话结束时，存储最终上下文：

def session_end():
    """在终止会话前调用。"""
    # 存储已完成的内容
    remember("context", "last_session_summary",
             f"Session ended at {time.now()}. Accomplished: ...",
             tags=["session"], priority="normal")
    
    # 更新任务状态
    for task in get_active_tasks():
        if task_in_progress(task):
            update_task(task["id"], description=f"In progress: {current_step}")

session_end()

完整模式

class PersistentAgent:
    def __init__(self):
        self.mind_key = os.environ["SYNAPSE_MIND_KEY"]
        self.url = "https://synapse.schaefer.zone"
    
    def run(self):
        # 1. 回放上下文
        context = self.session_start()
        
        # 2. 处理未读消息
        for msg in context["unread_messages"]:
            self.handle_message(msg)
        
        # 3. 恢复进行中的任务
        for task in context["active_tasks"]:
            self.continue_task(task)
        
        # 4. 做新工作
        self.do_work()
        
        # 5. 持久化状态
        self.session_end()

最佳实践

下一步

• LLM Cookbook — 实用模式
• 记忆最佳实践
• 多 Agent 协调