Session Management

Conversation lifecycle, compression, and memory extraction

Session manages the complete conversation lifecycle: message recording, context tracking, compression, and automatic memory extraction.

Overview

Session Lifecycle: Create → Interact → Commit

1. Create Session

   Initialize a new conversation session with unique ID

2. Add Messages

   Record user and assistant messages with multimodal content

3. Track Usage

   Record which contexts and skills were used

4. Commit

   Trigger compression and memory extraction

5. Compress

   Archive older messages, keep recent N rounds

6. Extract Memories

   Extract 6-category memories from conversation

7. Update Index

   Vectorize and index extracted memories

Core API

Session Creation

from openviking import OpenViking

client = OpenViking()

# Create or resume session
session = client.session(session_id="chat_001")

# Auto-generated session ID if not provided
session = client.session()  # Generates unique ID
print(session.session_id)   # e.g., "sess_abc123"

add_message

Add conversation messages with multimodal content:

# User message
await session.add_message(
    role="user",
    parts=[
        {"type": "text", "text": "How do I configure embedding?"}
    ]
)

# Assistant message with context references
await session.add_message(
    role="assistant",
    parts=[
        {"type": "text", "text": "Here's how to configure embedding..."},
        {
            "type": "context",
            "uri": "viking://resources/docs/config.md",
            "abstract": "Configuration guide for OpenViking"
        }
    ]
)

# Multimodal message
await session.add_message(
    role="user",
    parts=[
        {"type": "text", "text": "What's in this image?"},
        {"type": "image", "url": "file:///path/to/image.png"}
    ]
)

used

Record context and skill usage:

# Record used contexts (updates active_count)
await session.used(
    contexts=[
        "viking://resources/docs/config.md",
        "viking://user/memories/preferences/coding"
    ]
)

# Record used skill
await session.used(
    skill={
        "uri": "viking://agent/skills/code-search",
        "input": "search config files",
        "output": "found 3 configuration files",
        "success": True
    }
)

commit

Trigger compression and memory extraction:

result = await session.commit()

print(result)
# {
#   "status": "committed",
#   "memories_extracted": 5,
#   "active_count_updated": 2,
#   "archived": True
# }

Message Structure

Message

from dataclasses import dataclass
from datetime import datetime
from typing import List

@dataclass
class Message:
    id: str                  # msg_{UUID}
    role: str                # "user" | "assistant"
    parts: List[Part]        # Message parts (multimodal)
    created_at: datetime

Part Types

TextPart

@dataclass
class TextPart:
    type: str = "text"
    text: str            # Text content

# Usage
part = TextPart(text="Hello, how do I...")

ContextPart

@dataclass
class ContextPart:
    type: str = "context"
    uri: str             # Context URI
    abstract: str        # L0 abstract (for display)

# Usage
part = ContextPart(
    uri="viking://resources/docs/api.md",
    abstract="API documentation for OpenViking"
)

ToolPart

@dataclass
class ToolPart:
    type: str = "tool"
    tool_name: str       # Tool/skill name
    input: dict          # Tool input
    output: dict         # Tool output
    success: bool        # Whether successful

# Usage
part = ToolPart(
    tool_name="search-web",
    input={"query": "OpenViking"},
    output={"results": [...]},
    success=True
)

ImagePart

@dataclass
class ImagePart:
    type: str = "image"
    url: str             # Image URL or file path
    description: str     # Optional description

# Usage
part = ImagePart(
    url="file:///path/to/screenshot.png",
    description="Login page screenshot"
)

Compression Strategy

Sessions automatically compress when the message count exceeds a threshold, keeping recent rounds while archiving older history.

Archive Flow

Auto-archive triggered by commit():

1. Increment compression_index

   Track which compression cycle this is

   compression_index += 1  # e.g., 1, 2, 3, ...
   
2. Copy messages to archive

   Move older messages to archive directory

   archive_dir = f"viking://session/{session_id}/history/archive_{compression_index:03d}/"
   await copy_messages(current_messages, archive_dir)
   
3. Generate structured summary

   LLM generates summary of archived segment

   summary = await vlm.generate_summary(
       messages=archived_messages,
       format="structured"
   )
   
4. Write L0/L1 for archive

   Create abstract and overview for archived history

   await write_context(
       uri=archive_dir,
       abstract=extract_abstract(summary),
       overview=summary
   )
   
5. Clear current messages

   Keep only recent N rounds in active session

   current_messages = current_messages[-KEEP_RECENT_ROUNDS:]
   

Summary Format

# Session Summary

**One-line overview**: [Topic]: [Intent] | [Result] | [Status]

## Analysis
Key steps taken:
1. User asked about OAuth implementation
2. Retrieved OAuth 2.0 documentation
3. Provided code examples
4. User requested clarification on token refresh
5. Explained refresh token flow

## Primary Request and Intent
User's core goal: Implement OAuth 2.0 authentication in their API service

## Key Concepts
- OAuth 2.0 authorization flow
- Access tokens and refresh tokens
- Token validation and expiration

## Pending Tasks
- Implement token storage mechanism
- Add token refresh endpoint
- Test OAuth flow end-to-end

Memory Extraction

OpenViking automatically extracts 6 categories of memories from conversations, updating user and agent knowledge bases.

6 Memory Categories

User Memories

Category	Location	Mergeable	Description
profile	user/.overview.md	✅ Yes	User identity, attributes
preferences	user/memories/preferences/	✅ Yes	User preferences by topic
entities	user/memories/entities/	✅ Yes	People, projects, concepts
events	user/memories/events/	❌ No	Historical events, decisions

Agent Memories

Category	Location	Mergeable	Description
cases	agent/memories/cases/	❌ No	Problem + solution pairs
patterns	agent/memories/patterns/	✅ Yes	Reusable best practices

Extraction Flow

1. LLM Extract

   Extract candidate memories from conversation

   candidates = await memory_extractor.extract(
       messages=session_messages,
       categories=["profile", "preferences", "entities", ...]
   )
   
2. Vector Pre-filter

   Find similar existing memories using vector search

   for candidate in candidates:
       similar = await vector_index.search(
           query=candidate.content,
           target_uri=f"viking://user/memories/{candidate.category}/",
           limit=5
       )
   
3. LLM Dedup Decision

   LLM decides: skip, create, or merge

   decision = await deduplicator.decide(
       candidate=candidate,
       existing=similar_memories
   )
   # Returns: {"candidate": "create", "items": [{"action": "merge", "uri": ...}]}
   
4. Write to AGFS

   Execute dedup decision

   if decision.candidate == "create":
       await write_memory(candidate)
   elif decision.candidate == "skip":
       pass  # Duplicate, skip
   
5. Vectorize

   Index new/updated memories

   await embedding_queue.enqueue(memory_context)
   

Dedup Decisions

Candidate Decisions

Decision	Description
skip	Candidate is duplicate, skip and do nothing
create	Create candidate memory (optionally delete conflicting existing memories first)
none	Do not create candidate; resolve existing memories by item decisions

Item Decisions (Per-existing memory)

Decision	Description
merge	Merge candidate content into specified existing memory
delete	Delete specified conflicting existing memory

Example: Dedup Decision

# Candidate memory
candidate = {
    "category": "preferences",
    "topic": "coding",
    "content": "User prefers TypeScript over JavaScript and uses ESLint for linting"
}

# Existing similar memory
existing = {
    "uri": "viking://user/memories/preferences/coding",
    "content": "User prefers TypeScript over JavaScript"
}

# LLM dedup decision
decision = {
    "candidate": "create",  # Create new memory
    "items": [
        {
            "uri": "viking://user/memories/preferences/coding",
            "action": "merge"  # Merge into existing
        }
    ]
}

# Result: Existing memory updated to:
# "User prefers TypeScript over JavaScript and uses ESLint for linting"

Storage Structure

Session Directory

viking://session/{user_space}/{session_id}/
├── messages.jsonl            # Current messages (JSONL format)
├── .abstract.md              # L0: Current session summary
├── .overview.md              # L1: Current session overview
├── .meta.json                # Session metadata
├── history/                  # Archived history
│   ├── archive_001/
│   │   ├── messages.jsonl    # Archived messages
│   │   ├── .abstract.md      # Archive summary
│   │   └── .overview.md      # Archive overview
│   ├── archive_002/
│   └── archive_NNN/
└── tools/                    # Tool executions
    └── {tool_id}/
        └── tool.json         # Tool call record

Memory Directories

viking://user/{user_space}/memories/
├── .overview.md              # User profile (appendable)
├── preferences/              # User preferences
│   ├── coding/               # Topic-based organization
│   ├── communication/
│   └── tools/
├── entities/                 # Entity memories
│   ├── project_openviking/
│   ├── colleague_alice/
│   └── concept_rag/
└── events/                   # Event records
    ├── 2024-01-15_decided_refactor/
    └── 2024-02-20_completed_feature/

viking://agent/{agent_space}/memories/
├── cases/                    # Specific problem-solution pairs
│   ├── debug_import_error/
│   └── fix_api_timeout/
└── patterns/                 # Reusable patterns
    ├── debugging_workflow/
    └── code_review_checklist/

Complete Example

from openviking import OpenViking

client = OpenViking()

# Create session
session = client.session(session_id="oauth_help")

# User asks question
await session.add_message(
    "user",
    [{"type": "text", "text": "I prefer using TypeScript for all my projects. How do I implement OAuth?"}]
)

# Search for relevant contexts
results = await client.search(
    "OAuth implementation in TypeScript",
    session_info=session
)

# Record used contexts
await session.used(
    contexts=[ctx.uri for ctx in results.resources[:3]]
)

# Assistant responds
await session.add_message(
    "assistant",
    [
        {"type": "text", "text": "Here's how to implement OAuth in TypeScript..."},
        {"type": "context", "uri": results.resources[0].uri, "abstract": results.resources[0].abstract}
    ]
)

# Commit session (triggers compression + memory extraction)
result = await session.commit()

print(f"Extracted {result['memories_extracted']} memories")
# Automatically extracted:
# - preference: "User prefers TypeScript for projects"
# - entity: "OAuth implementation project"
# - case: "OAuth implementation in TypeScript"

# Later: Search user preferences
prefs = await client.find(
    "language preferences",
    target_uri="viking://user/memories/preferences/"
)

for pref in prefs.memories:
    overview = await client.overview(pref.uri)
    print(f"Preference: {overview}")
    # Output: "User prefers TypeScript for all projects..."

Related Concepts

Architecture

System architecture and data flow

Context Types

6 memory categories explained

Extraction

Memory extraction pipeline

Retrieval

How memories are searched