Manual Thread Management

Manually control threads for multi-user applications.

While agents automatically manage threads, you can also control threads manually for multi-user applications or complex workflows.

Creating and Switching Threads

You can create multiple threads to handle different conversations simultaneously. By using create_thread with metadata, you can tag threads with user IDs or session info. The use_thread method then lets you switch the active context, ensuring that new messages go to the correct conversation.

from peargent import create_history
from peargent.storage import Sqlite

history = create_history(store_type=Sqlite(database_path="./app.db"))

# Create threads for different users
alice_thread = history.create_thread(metadata={"user_id": "alice", "session": "web"})
bob_thread = history.create_thread(metadata={"user_id": "bob", "session": "mobile"})

# Switch between threads
history.use_thread(alice_thread)
history.add_user_message("What's the weather?")

history.use_thread(bob_thread)
history.add_user_message("Show my orders")

# List all threads
all_threads = history.list_threads()
print(f"Total threads: {len(all_threads)}")

# Get thread with metadata
thread = history.get_thread(alice_thread)
print(f"User: {thread.metadata.get('user_id')}")
print(f"Messages: {len(thread.messages)}")

Multi-User Application Pattern

For applications serving multiple users, you need to ensure each user gets their own conversation history. This pattern shows how to use metadata to look up an existing thread for a user. If a thread is found, the agent resumes that conversation; if not, a new thread is created. This allows a single agent instance to handle many users concurrently.

from peargent import create_agent, create_history, HistoryConfig
from peargent.storage import Postgresql
from peargent.models import openai

# Shared history store for all users
history = create_history(
    store_type=Postgresql(
        connection_string="postgresql://user:pass@localhost/app_db"
    )
)

# Create agent (reused across users)
agent = create_agent(
    name="Assistant",
    description="Customer support assistant",
    persona="You are a helpful customer support agent.",
    model=openai("gpt-4o")
)

def handle_user_message(user_id: str, message: str):
    """Handle message from a specific user."""

    # Find or create thread for this user
    all_threads = history.list_threads()
    user_thread = None

    for thread_id in all_threads:
        thread = history.get_thread(thread_id)
        if thread.metadata.get("user_id") == user_id:
            user_thread = thread_id
            break

    if not user_thread:
        # Create new thread for this user
        user_thread = history.create_thread(metadata={"user_id": user_id})

    # Set active thread
    history.use_thread(user_thread)

    # Add user message
    history.add_user_message(message)

    # Get response from agent
    # Note: Agent needs to load this history manually or use temporary_memory
    response = agent.run(message)

    # Add assistant response
    history.add_assistant_message(response, agent="Assistant")

    return response

# Usage
response1 = handle_user_message("alice", "What's my order status?")
response2 = handle_user_message("bob", "I need help with returns")
response3 = handle_user_message("alice", "Thanks!")  # Same thread as first message

Manual Thread Management

Manually control threads for multi-user applications.

Creating and Switching Threads

Multi-User Application Pattern

On this page