Conversation Summarization

class ConversationSummary(BaseModel):
    chat_id: str
    summary: str | None  # 1-2 sentence description
    request: str | None  # User's overall request
    topic: str | None  # Deprecated field
    languages: list[str] | None  # Human and programming languages
    task: str | None  # Task being performed
    concerning_score: int | None  # Safety score (1-5)
    user_frustration: int | None  # Frustration level (1-5)
    assistant_errors: list[str] | None  # List of assistant mistakes
    metadata: dict  # Original metadata + custom fields

from kura.summarisation import summarise_conversations, SummaryModel

model = SummaryModel(
    model="openai/gpt-4o-mini",
    max_concurrent_requests=50
)

summaries = await summarise_conversations(
    conversations=conversations,
    model=model
)

print(summaries[0].summary)
# Output: "User asks for help debugging a Python pandas DataFrame indexing error"

from kura.checkpoints import JSONLCheckpointManager

checkpoint_mgr = JSONLCheckpointManager("./checkpoints")

summaries = await summarise_conversations(
    conversations=conversations,
    model=model,
    checkpoint_manager=checkpoint_mgr
)

# On subsequent runs, summaries are loaded from checkpoint
# without re-calling the LLM

class SummaryModel(BaseSummaryModel):
    def __init__(
        self,
        model: Union[str, "KnownModelName"] = "openai/gpt-4o-mini",
        max_concurrent_requests: int = 50,
        checkpoint_filename: str = "summaries",
        console: Optional[Console] = None,
        cache: Optional[CacheStrategy] = None,
    )

"User requests assistance creating a React component for a dropdown menu with custom styling"

"The user's overall request for the assistant is to help build a reusable dropdown component in React"

["english", "react", "javascript", "css"]

"The task is to generate code for a React component with styling"

[
    "Provided incomplete code example",
    "Failed to address styling requirements",
    "Ignored user's preference for functional components"
]

custom_prompt = """
Analyze this technical support conversation and extract:
1. The user's technical problem
2. Steps taken to resolve it
3. Whether the issue was resolved
4. Customer satisfaction level (1-5)

Conversation:
{% for message in conversation.messages %}
{{ message.role }}: {{ message.content }}
{% endfor %}
"""

summaries = await summarise_conversations(
    conversations=conversations,
    model=model,
    prompt=custom_prompt
)

from kura.types.summarisation import GeneratedSummary

class DetailedSummary(GeneratedSummary):
    sentiment: str  # "positive", "negative", "neutral"
    technical_complexity: int  # 1-10 scale
    product_area: str  # Which product feature was discussed

summaries = await summarise_conversations(
    conversations=conversations,
    model=model,
    response_schema=DetailedSummary
)

# Custom fields are available in metadata
print(summaries[0].metadata["sentiment"])  # "positive"
print(summaries[0].metadata["technical_complexity"])  # 7

from kura.cache import DiskCache

cache = DiskCache(
    cache_dir="./cache",
    ttl=86400 * 7  # 7 days
)

model = SummaryModel(
    model="openai/gpt-4o-mini",
    cache=cache
)

# First run: Calls LLM for all conversations
summaries = await summarise_conversations(
    conversations=conversations,
    model=model
)

# Second run: Loads from cache instantly
summaries = await summarise_conversations(
    conversations=conversations,
    model=model
)

from rich.console import Console

console = Console()

model = SummaryModel(
    model="openai/gpt-4o-mini",
    console=console
)

summaries = await summarise_conversations(
    conversations=conversations,
    model=model
)

from kura.summarisation import USAGE_ANALYSIS_PROMPT

summaries = await summarise_conversations(
    conversations=conversations,
    model=model,
    prompt=USAGE_ANALYSIS_PROMPT
)

# From kura/summarisation.py:257
self.semaphore = asyncio.Semaphore(self.max_concurrent_requests)

# From kura/summarisation.py:332
async with self.semaphore:
    resp = await client.chat.completions.create(...)

# Use checkpointing to avoid re-analysis
checkpoint_mgr = JSONLCheckpointManager("./checkpoints")

# Use caching for duplicate conversations
cache = DiskCache("./cache")

# Process in batches for large datasets
for batch in batched(conversations, 1000):
    summaries = await summarise_conversations(
        conversations=batch,
        model=model,
        checkpoint_manager=checkpoint_mgr
    )

# Adjust concurrency based on provider
model = SummaryModel(
    model="openai/gpt-4o-mini",
    max_concurrent_requests=50  # OpenAI: 3,500 RPM
)

model = SummaryModel(
    model="anthropic/claude-3-5-sonnet",
    max_concurrent_requests=5  # Anthropic: lower default limits
)