{"title":"构建持久化 LLM Agent","slug":"persistent-llm-agent","category":"guides","summary":"使用 Synapse 构建跨会话记忆的 LLM Agent 的分步指南。","audience":["human","llm"],"tags":["guide","agent","persistence","architecture"],"difficulty":"intermediate","updated":"2026-06-27","word_count":103,"read_minutes":1,"lang":"zh","translated":true,"requested_lang":"zh","content_markdown":"\n# 概览\n\n本指南带你构建一个使用 Synapse 跨会话持久化上下文的 LLM Agent。完成时，你的 Agent 将能够：\n\n- 会话开始时回放过往上下文\n- 随时存储新学到的内容\n- 跨会话跟踪多步任务\n- 通过异步聊天与人类通信\n\n## 架构\n\n```text\n┌──────────────┐ 回放/存储 ┌──────────┐\n│ LLM Agent │ ◀──────────────▶ │ Synapse │\n│ (你的代码) │ │ API │\n└──────────────┘ └──────────┘\n │\n │ 轮询/回复\n ▼\n┌──────────────┐\n│ 人类 │ (浏览器或聊天 UI)\n└──────────────┘\n```\n\n## 第 1 步：设置 Mind Key\n\n```bash\n# 注册并获取 JWT\nJWT=$(curl -s -X POST https://synapse.schaefer.zone/register \\\n -H \"Content-Type: application/json\" \\\n -d '{\"email\":\"agent@example.com\",\"password\":\"secret\"}' | jq -r .jwt)\n\n# 创建 Mind 并获取 Mind Key\nMIND_KEY=$(curl -s -X POST https://synapse.schaefer.zone/minds \\\n -H \"Authorization: Bearer $JWT\" \\\n -H \"Content-Type: application/json\" \\\n -d '{\"name\":\"persistent-agent\",\"description\":\"My persistent agent\"}' | jq -r .mind_key)\n\necho \"Save this: $MIND_KEY\"\n```\n\n## 第 2 步：会话启动协议\n\n每次会话开始时，回放所有记忆：\n\n```python\nimport os\nimport requests\n\nMIND_KEY = os.environ[\"SYNAPSE_MIND_KEY\"]\nURL = \"https://synapse.schaefer.zone\"\n\ndef session_start():\n \"\"\"每次会话开始时调用。\"\"\"\n # 1. 回放所有记忆\n r = requests.get(\n f\"{URL}/memory/recall\",\n headers={\"Authorization\": f\"Bearer {MIND_KEY}\"}\n )\n memories = r.text # 纯文本摘要\n \n # 2. 检查未读聊天消息\n r = requests.get(\n f\"{URL}/chat/poll\",\n headers={\"Authorization\": f\"Bearer {MIND_KEY}\"}\n )\n messages = r.json().get(\"messages\", [])\n \n # 3. 检查进行中的任务\n r = requests.get(\n f\"{URL}/mind/tasks?status=in_progress\",\n headers={\"Authorization\": f\"Bearer {MIND_KEY}\"}\n )\n tasks = r.json().get(\"tasks\", [])\n \n return {\n \"memories\": memories,\n \"unread_messages\": messages,\n \"active_tasks\": tasks,\n }\n\ncontext = session_start()\n# 用此上下文构建系统 prompt\n```\n\n## 第 3 步：存储新学习内容\n\n每当 Agent 学到值得记住的内容：\n\n```python\ndef remember(category, key, content, tags=None, priority=\"normal\"):\n \"\"\"存储一条记忆。\"\"\"\n requests.post(\n f\"{URL}/memory\",\n headers={\n \"Authorization\": f\"Bearer {MIND_KEY}\",\n \"Content-Type\": \"application/json\",\n },\n json={\n \"category\": category,\n \"key\": key,\n \"content\": content,\n \"tags\": tags or [],\n \"priority\": priority,\n }\n )\n\n# 示例\nremember(\"identity\", \"user_name\", \"User is Michael Schäfer\", \n tags=[\"person\"], priority=\"critical\")\nremember(\"preference\", \"communication_style\", \n \"User prefers concise technical responses\",\n tags=[\"communication\"])\nremember(\"project\", \"current_project\", \n \"Building Synapse v1.6.0 with docs system\",\n tags=[\"synapse\", \"docs\"], priority=\"high\")\nremember(\"mistake\", \"npm_version_bump\", \n \"Always bump package.json version after changes\",\n tags=[\"npm\", \"ci\"], priority=\"high\")\n```\n\n## 第 4 步：任务管理\n\n跨会话跟踪多步工作：\n\n```python\ndef create_task(title, description=\"\", priority=\"normal\"):\n r = requests.post(\n f\"{URL}/mind/task\",\n headers={\"Authorization\": f\"Bearer {MIND_KEY}\",\n \"Content-Type\": \"application/json\"},\n json={\"title\": title, \"description\": description, \"priority\": priority}\n )\n return r.json()[\"id\"]\n\ndef update_task(task_id, status=None, description=None):\n payload = {}\n if status: payload[\"status\"] = status\n if description: payload[\"description\"] = description\n requests.put(\n f\"{URL}/mind/task/{task_id}\",\n headers={\"Authorization\": f\"Bearer {MIND_KEY}\",\n \"Content-Type\": \"application/json\"},\n json=payload\n )\n\n# 跨多会话工作流\ntask_id = create_task(\"Deploy v1.6.0\", \"Push docs system to production\", \"high\")\nupdate_task(task_id, status=\"in_progress\")\n# ... 跨多个会话工作 ...\nupdate_task(task_id, status=\"done\")\n```\n\n## 第 5 步：与人类异步聊天\n\n在工具调用之间轮询消息：\n\n```python\nimport time\n\ndef poll_messages():\n r = requests.get(\n f\"{URL}/chat/poll\",\n headers={\"Authorization\": f\"Bearer {MIND_KEY}\"}\n )\n return r.json().get(\"messages\", [])\n\ndef reply(content):\n requests.post(\n f\"{URL}/chat/reply\",\n headers={\"Authorization\": f\"Bearer {MIND_KEY}\",\n \"Content-Type\": \"application/json\"},\n json={\"content\": content}\n )\n\n# 主循环\nwhile working:\n # 轮询人类消息\n for msg in poll_messages():\n print(f\"Human: {msg['content']}\")\n reply(f\"Got it: {msg['content']}. Working on it.\")\n \n # 执行一个工作单元\n do_work()\n \n time.sleep(30) # 不要轮询过于频繁\n```\n\n## 第 6 步：会话结束协议\n\n会话结束时，存储最终上下文：\n\n```python\ndef session_end():\n \"\"\"在终止会话前调用。\"\"\"\n # 存储已完成的内容\n remember(\"context\", \"last_session_summary\",\n f\"Session ended at {time.now()}. Accomplished: ...\",\n tags=[\"session\"], priority=\"normal\")\n \n # 更新任务状态\n for task in get_active_tasks():\n if task_in_progress(task):\n update_task(task[\"id\"], description=f\"In progress: {current_step}\")\n\nsession_end()\n```\n\n## 完整模式\n\n```python\nclass PersistentAgent:\n def __init__(self):\n self.mind_key = os.environ[\"SYNAPSE_MIND_KEY\"]\n self.url = \"https://synapse.schaefer.zone\"\n \n def run(self):\n # 1. 回放上下文\n context = self.session_start()\n \n # 2. 处理未读消息\n for msg in context[\"unread_messages\"]:\n self.handle_message(msg)\n \n # 3. 恢复进行中的任务\n for task in context[\"active_tasks\"]:\n self.continue_task(task)\n \n # 4. 做新工作\n self.do_work()\n \n # 5. 持久化状态\n self.session_end()\n```\n\n## 最佳实践\n\n> [!TIP]\n>\n> - **始终先回放** — 不加载上下文就不要开始工作\n> - **主动存储** — 不要等到会话结束\n> - **使用有意义的 key** — `user_name`、`project_status`，而非 `mem_001`\n> - **给一切打标签** — 标签支撑搜索与过滤\n> - **设定合理的优先级** — 不是所有事都 `critical`\n\n## 下一步\n\n- [LLM Cookbook](/docs/llm-cookbook/session-start-pattern) — 实用模式\n- [记忆最佳实践](/docs/guides/memory-best-practices)\n- [多 Agent 协调](/docs/guides/multi-agent-coordination)\n","content_html":"

概览

\n

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

\n\n

架构

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

第 1 步：设置 Mind Key

\n
# 注册并获取 JWT\nJWT=$(curl -s -X POST https://synapse.schaefer.zone/register \\\n  -H "Content-Type: application/json" \\\n  -d '{"email":"agent@example.com","password":"secret"}' | jq -r .jwt)\n\n# 创建 Mind 并获取 Mind Key\nMIND_KEY=$(curl -s -X POST https://synapse.schaefer.zone/minds \\\n  -H "Authorization: Bearer $JWT" \\\n  -H "Content-Type: application/json" \\\n  -d '{"name":"persistent-agent","description":"My persistent agent"}' | jq -r .mind_key)\n\necho "Save this: $MIND_KEY"

第 2 步：会话启动协议

\n

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

\n
import os\nimport requests\n\nMIND_KEY = os.environ["SYNAPSE_MIND_KEY"]\nURL = "https://synapse.schaefer.zone"\n\ndef session_start():\n    """每次会话开始时调用。"""\n    # 1. 回放所有记忆\n    r = requests.get(\n        f"{URL}/memory/recall",\n        headers={"Authorization": f"Bearer {MIND_KEY}"}\n    )\n    memories = r.text  # 纯文本摘要\n    \n    # 2. 检查未读聊天消息\n    r = requests.get(\n        f"{URL}/chat/poll",\n        headers={"Authorization": f"Bearer {MIND_KEY}"}\n    )\n    messages = r.json().get("messages", [])\n    \n    # 3. 检查进行中的任务\n    r = requests.get(\n        f"{URL}/mind/tasks?status=in_progress",\n        headers={"Authorization": f"Bearer {MIND_KEY}"}\n    )\n    tasks = r.json().get("tasks", [])\n    \n    return {\n        "memories": memories,\n        "unread_messages": messages,\n        "active_tasks": tasks,\n    }\n\ncontext = session_start()\n# 用此上下文构建系统 prompt

第 3 步：存储新学习内容

\n

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

\n
def remember(category, key, content, tags=None, priority="normal"):\n    """存储一条记忆。"""\n    requests.post(\n        f"{URL}/memory",\n        headers={\n            "Authorization": f"Bearer {MIND_KEY}",\n            "Content-Type": "application/json",\n        },\n        json={\n            "category": category,\n            "key": key,\n            "content": content,\n            "tags": tags or [],\n            "priority": priority,\n        }\n    )\n\n# 示例\nremember("identity", "user_name", "User is Michael Schäfer", \n         tags=["person"], priority="critical")\nremember("preference", "communication_style", \n         "User prefers concise technical responses",\n         tags=["communication"])\nremember("project", "current_project", \n         "Building Synapse v1.6.0 with docs system",\n         tags=["synapse", "docs"], priority="high")\nremember("mistake", "npm_version_bump", \n         "Always bump package.json version after changes",\n         tags=["npm", "ci"], priority="high")

第 4 步：任务管理

\n

跨会话跟踪多步工作：

\n
def create_task(title, description="", priority="normal"):\n    r = requests.post(\n        f"{URL}/mind/task",\n        headers={"Authorization": f"Bearer {MIND_KEY}",\n                 "Content-Type": "application/json"},\n        json={"title": title, "description": description, "priority": priority}\n    )\n    return r.json()["id"]\n\ndef update_task(task_id, status=None, description=None):\n    payload = {}\n    if status: payload["status"] = status\n    if description: payload["description"] = description\n    requests.put(\n        f"{URL}/mind/task/{task_id}",\n        headers={"Authorization": f"Bearer {MIND_KEY}",\n                 "Content-Type": "application/json"},\n        json=payload\n    )\n\n# 跨多会话工作流\ntask_id = create_task("Deploy v1.6.0", "Push docs system to production", "high")\nupdate_task(task_id, status="in_progress")\n# ... 跨多个会话工作 ...\nupdate_task(task_id, status="done")

第 5 步：与人类异步聊天

\n

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

\n
import time\n\ndef poll_messages():\n    r = requests.get(\n        f"{URL}/chat/poll",\n        headers={"Authorization": f"Bearer {MIND_KEY}"}\n    )\n    return r.json().get("messages", [])\n\ndef reply(content):\n    requests.post(\n        f"{URL}/chat/reply",\n        headers={"Authorization": f"Bearer {MIND_KEY}",\n                 "Content-Type": "application/json"},\n        json={"content": content}\n    )\n\n# 主循环\nwhile working:\n    # 轮询人类消息\n    for msg in poll_messages():\n        print(f"Human: {msg['content']}")\n        reply(f"Got it: {msg['content']}. Working on it.")\n    \n    # 执行一个工作单元\n    do_work()\n    \n    time.sleep(30)  # 不要轮询过于频繁

第 6 步：会话结束协议

\n

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

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

完整模式

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

最佳实践

\n

下一步

\n\n","urls":{"html":"/docs/guides/persistent-llm-agent","text":"/docs/guides/persistent-llm-agent?format=text","json":"/docs/guides/persistent-llm-agent?format=json","llm":"/docs/guides/persistent-llm-agent?format=llm"},"translations_available":["en","zh","hi","es","fr","ar","pt","ru","ja","de","it","ko","nl","pl","tr","sv","vi","th","id","uk"]}