Xây dựng LLM Agent cố định

intermediate 🤖 Thân thiện với LLM 1 phút · 273 từ Cập nhật: 2026-06-27 Tiếng Việt

Hướng dẫn từng bước xây dựng LLM agent ghi nhớ qua các phiên sử dụng Synapse.

Đọc bài viết này ở định dạng: 📄 Văn bản thuần ⚡ JSON 🤖 Định dạng LLM

Tổng quan

Hướng dẫn này dẫn qua việc xây dựng LLM agent lưu ngữ cảnh qua các phiên sử dụng Synapse. Cuối cùng, agent của bạn sẽ:

Thu hồi ngữ cảnh quá khứ ở đầu phiên
Lưu học hỏi mới khi chúng xảy ra
Theo dõi tác vụ nhiều bước qua các phiên
Giao tiếp với con người qua chat bất đồng bộ

Kiến trúc

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

Bước 1: Thiết lập 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"

Bước 2: Giao thức bắt đầu phiên

Ở đầu mỗi phiên, thu hồi tất cả bộ nhớ:

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

Bước 3: Lưu học hỏi mới

Bất cứ khi nào agent học được điều gì đáng nhớ:

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

Bước 4: Quản lý tác vụ

Theo dõi công việc nhiều bước qua các phiên:

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

Bước 5: Chat bất đồng bộ với con người

Poll tin nhắn giữa các lệnh gọi công cụ:

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

Bước 6: Giao thức kết thúc phiên

Ở cuối phiên, lưu ngữ cảnh cuối:

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

Mẫu hoàn chỉnh

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