Skip to main content

Membangun Agen LLM Persisten

Panduan langkah demi langkah untuk membangun agen LLM yang mengingat lintas sesi menggunakan Synapse.


Ikhtisar

Panduan ini memandu membangun agen LLM yang mempertahankan konteks lintas sesi menggunakan Synapse. Pada akhirnya, agen Anda akan:

  • Recall konteks masa lalu di awal sesi
  • Menyimpan pembelajaran baru saat terjadi
  • Melacak tugas multi-langkah lintas sesi
  • Berkomunikasi dengan manusia melalui chat asinkron

Arsitektur

┌──────────────┐   recall/store   ┌──────────┐
│  LLM Agent   │ ◀──────────────▶ │ Synapse  │
│ (your code)  │                  │   API    │
└──────────────┘                  └──────────┘
       │
       │ poll/reply
       ▼
┌──────────────┐
│    Human     │ (browser or chat UI)
└──────────────┘

Langkah 1: Siapkan Mind Key

# Register and get 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)

# Create mind and get 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"

Langkah 2: Protokol Awal Sesi

Di awal setiap sesi, recall semua memori:

import os
import requests

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

def session_start():
    """Call this at the start of every session."""
    # 1. Recall all memories
    r = requests.get(
        f"{URL}/memory/recall",
        headers={"Authorization": f"Bearer {MIND_KEY}"}
    )
    memories = r.text  # plain text summary
    
    # 2. Check for unread chat messages
    r = requests.get(
        f"{URL}/chat/poll",
        headers={"Authorization": f"Bearer {MIND_KEY}"}
    )
    messages = r.json().get("messages", [])
    
    # 3. Check in-progress tasks
    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()
# Build system prompt with this context

Langkah 3: Simpan Pembelajaran Baru

Setiap kali agen mempelajari sesuatu yang patut diingat:

def remember(category, key, content, tags=None, priority="normal"):
    """Store a memory."""
    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,
        }
    )

# Examples
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")

Langkah 4: Manajemen Tugas

Lacak pekerjaan multi-langkah lintas sesi:

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
    )

# Multi-session workflow
task_id = create_task("Deploy v1.6.0", "Push docs system to production", "high")
update_task(task_id, status="in_progress")
# ... work across multiple sessions ...
update_task(task_id, status="done")

Langkah 5: Chat Asinkron dengan Manusia

Polling pesan di antara pemanggilan alat:

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}
    )

# Main loop
while working:
    # Poll for human messages
    for msg in poll_messages():
        print(f"Human: {msg['content']}")
        reply(f"Got it: {msg['content']}. Working on it.")
    
    # Do one unit of work
    do_work()
    
    time.sleep(30)  # don't poll too frequently

Langkah 6: Protokol Akhir Sesi

Di akhir sesi, simpan konteks akhir:

def session_end():
    """Call this before terminating the session."""
    # Store what we accomplished
    remember("context", "last_session_summary",
             f"Session ended at {time.now()}. Accomplished: ...",
             tags=["session"], priority="normal")
    
    # Update task statuses
    for task in get_active_tasks():
        if task_in_progress(task):
            update_task(task["id"], description=f"In progress: {current_step}")

session_end()

Pola Lengkap

class PersistentAgent:
    def __init__(self):
        self.mind_key = os.environ["SYNAPSE_MIND_KEY"]
        self.url = "https://synapse.schaefer.zone"
    
    def run(self):
        # 1. Recall context
        context = self.session_start()
        
        # 2. Process unread messages
        for msg in context["unread_messages"]:
            self.handle_message(msg)
        
        # 3. Resume active tasks
        for task in context["active_tasks"]:
            self.continue_task(task)
        
        # 4. Do new work
        self.do_work()
        
        # 5. Persist state
        self.session_end()

Praktik Terbaik

Langkah Berikutnya