feat: phase 3: introspective meta-dreaming
This commit is contained in:
Benjamin McCormick
parent
5ac7b85ef7
commit
97f3409280
|
|
@ -0,0 +1,454 @@
|
|||
"""
|
||||
Introspection module for meta-cognitive analysis of Honcho's performance.
|
||||
|
||||
This module provides functionality to analyze how Honcho is performing for a workspace
|
||||
and generate configuration suggestions based on observed patterns.
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import datetime
|
||||
import json
|
||||
import logging
|
||||
|
||||
from sqlalchemy import func, select
|
||||
from sqlalchemy.ext.asyncio import AsyncSession
|
||||
|
||||
from src import crud, models
|
||||
from src.config import settings
|
||||
from src.schemas import (
|
||||
IntrospectionReport,
|
||||
IntrospectionSignals,
|
||||
IntrospectionSuggestion,
|
||||
)
|
||||
from src.utils.clients import honcho_llm_call
|
||||
|
||||
logger = logging.getLogger(__name__)
|
||||
|
||||
|
||||
# Reserved peer names for storing introspection reports
|
||||
SYSTEM_OBSERVER = "_system"
|
||||
INTROSPECTION_OBSERVED = "_introspection"
|
||||
|
||||
|
||||
async def gather_introspection_context(
|
||||
db: AsyncSession,
|
||||
workspace_name: str,
|
||||
) -> IntrospectionSignals:
|
||||
"""
|
||||
Gather performance signals from a workspace for introspection analysis.
|
||||
|
||||
Args:
|
||||
db: Database session
|
||||
workspace_name: Name of the workspace to analyze
|
||||
|
||||
Returns:
|
||||
IntrospectionSignals containing performance metrics and configuration
|
||||
"""
|
||||
# Calculate time window (last 30 days)
|
||||
since = datetime.datetime.now(datetime.timezone.utc) - datetime.timedelta(days=30)
|
||||
|
||||
# Get dialectic trace statistics
|
||||
try:
|
||||
trace_stats = await crud.get_dialectic_trace_stats(
|
||||
db, workspace_name, since=since
|
||||
)
|
||||
except Exception as e:
|
||||
logger.warning(f"Failed to get dialectic trace stats: {e}")
|
||||
trace_stats = {
|
||||
"total_queries": 0,
|
||||
"avg_duration_ms": 0.0,
|
||||
"abstention_count": 0,
|
||||
"abstention_rate": 0.0,
|
||||
}
|
||||
|
||||
# Get recent dialectic queries for sample analysis
|
||||
recent_queries: list[str] = []
|
||||
try:
|
||||
traces = await crud.get_dialectic_traces(db, workspace_name, limit=20)
|
||||
recent_queries = [t.query for t in traces]
|
||||
except Exception as e:
|
||||
logger.warning(f"Failed to get recent dialectic traces: {e}")
|
||||
|
||||
# Get observation counts by level
|
||||
observations_by_level: dict[str, int] = {}
|
||||
total_observations = 0
|
||||
contradiction_count = 0
|
||||
try:
|
||||
# Query documents grouped by level
|
||||
stmt = (
|
||||
select(
|
||||
models.Document.level,
|
||||
func.count(models.Document.id).label("count"),
|
||||
)
|
||||
.where(models.Document.workspace_name == workspace_name)
|
||||
.where(models.Document.deleted_at.is_(None))
|
||||
.group_by(models.Document.level)
|
||||
)
|
||||
result = await db.execute(stmt)
|
||||
for row in result.all():
|
||||
level, count = row
|
||||
observations_by_level[level] = count
|
||||
total_observations += count
|
||||
if level == "contradiction":
|
||||
contradiction_count = count
|
||||
except Exception as e:
|
||||
logger.warning(f"Failed to get observation counts: {e}")
|
||||
|
||||
# Get peer count
|
||||
total_peers = 0
|
||||
try:
|
||||
stmt = (
|
||||
select(func.count(models.Peer.id))
|
||||
.where(models.Peer.workspace_name == workspace_name)
|
||||
.where(~models.Peer.name.startswith("_")) # Exclude system peers
|
||||
)
|
||||
result = await db.execute(stmt)
|
||||
total_peers = result.scalar() or 0
|
||||
except Exception as e:
|
||||
logger.warning(f"Failed to get peer count: {e}")
|
||||
|
||||
# Get session count
|
||||
total_sessions = 0
|
||||
try:
|
||||
stmt = select(func.count(models.Session.id)).where(
|
||||
models.Session.workspace_name == workspace_name
|
||||
)
|
||||
result = await db.execute(stmt)
|
||||
total_sessions = result.scalar() or 0
|
||||
except Exception as e:
|
||||
logger.warning(f"Failed to get session count: {e}")
|
||||
|
||||
# Get current agent configuration
|
||||
current_deriver_rules = ""
|
||||
current_dialectic_rules = ""
|
||||
try:
|
||||
agent_config = await crud.get_workspace_agent_config(db, workspace_name)
|
||||
current_deriver_rules = agent_config.deriver_rules
|
||||
current_dialectic_rules = agent_config.dialectic_rules
|
||||
except Exception as e:
|
||||
logger.warning(f"Failed to get workspace agent config: {e}")
|
||||
|
||||
return IntrospectionSignals(
|
||||
total_dialectic_queries=int(trace_stats["total_queries"]),
|
||||
avg_dialectic_duration_ms=float(trace_stats["avg_duration_ms"]),
|
||||
abstention_count=int(trace_stats["abstention_count"]),
|
||||
abstention_rate=float(trace_stats["abstention_rate"]),
|
||||
recent_queries=recent_queries,
|
||||
total_observations=total_observations,
|
||||
observations_by_level=observations_by_level,
|
||||
contradiction_count=contradiction_count,
|
||||
total_peers=total_peers,
|
||||
total_sessions=total_sessions,
|
||||
current_deriver_rules=current_deriver_rules,
|
||||
current_dialectic_rules=current_dialectic_rules,
|
||||
)
|
||||
|
||||
|
||||
def build_introspection_prompt(signals: IntrospectionSignals) -> str:
|
||||
"""
|
||||
Build a prompt for the LLM to analyze workspace signals and suggest improvements.
|
||||
|
||||
Args:
|
||||
signals: Performance signals gathered from the workspace
|
||||
|
||||
Returns:
|
||||
A formatted prompt string for the LLM
|
||||
"""
|
||||
queries_sample = (
|
||||
"\n".join(f" - {q}" for q in signals.recent_queries[:10])
|
||||
if signals.recent_queries
|
||||
else " (No recent queries)"
|
||||
)
|
||||
|
||||
observations_breakdown = (
|
||||
"\n".join(
|
||||
f" - {level}: {count}"
|
||||
for level, count in signals.observations_by_level.items()
|
||||
)
|
||||
if signals.observations_by_level
|
||||
else " (No observations)"
|
||||
)
|
||||
|
||||
return f"""You are analyzing the performance of a Honcho workspace to identify issues and suggest configuration improvements.
|
||||
|
||||
## Workspace Performance Signals
|
||||
|
||||
### Dialectic API Usage (Last 30 Days)
|
||||
- Total queries: {signals.total_dialectic_queries}
|
||||
- Average response time: {signals.avg_dialectic_duration_ms:.1f}ms
|
||||
- Abstention count: {signals.abstention_count}
|
||||
- Abstention rate: {signals.abstention_rate:.1%}
|
||||
|
||||
### Recent Query Samples
|
||||
{queries_sample}
|
||||
|
||||
### Memory Statistics
|
||||
- Total observations: {signals.total_observations}
|
||||
- Observations by level:
|
||||
{observations_breakdown}
|
||||
- Contradictions detected: {signals.contradiction_count}
|
||||
|
||||
### Workspace Scale
|
||||
- Total peers: {signals.total_peers}
|
||||
- Total sessions: {signals.total_sessions}
|
||||
|
||||
### Current Configuration
|
||||
Deriver rules:
|
||||
```
|
||||
{signals.current_deriver_rules or "(empty - using defaults)"}
|
||||
```
|
||||
|
||||
Dialectic rules:
|
||||
```
|
||||
{signals.current_dialectic_rules or "(empty - using defaults)"}
|
||||
```
|
||||
|
||||
## Your Task
|
||||
|
||||
1. **Analyze the application type**: Based on the query samples and usage patterns, what kind of application is this workspace likely supporting? (e.g., customer support, personal assistant, educational tool, etc.)
|
||||
|
||||
2. **Identify performance issues**: Look for:
|
||||
- High abstention rate (>30% suggests memory gaps or query-memory mismatch)
|
||||
- High contradiction count (suggests conflicting information being stored)
|
||||
- Low observation count relative to query volume (suggests underutilization of memory)
|
||||
- Query patterns that suggest certain topics aren't being captured
|
||||
|
||||
3. **Suggest specific rule changes**: For each suggestion, specify:
|
||||
- Whether it's for `deriver_rules` or `dialectic_rules`
|
||||
- The exact suggested rule text
|
||||
- Why this change would help
|
||||
|
||||
Respond with a JSON object matching this schema:
|
||||
{{
|
||||
"performance_summary": "A 2-3 sentence summary of the workspace's performance",
|
||||
"identified_issues": ["Issue 1", "Issue 2", ...],
|
||||
"suggestions": [
|
||||
{{
|
||||
"target": "deriver_rules" | "dialectic_rules",
|
||||
"current_value": "current rule text",
|
||||
"suggested_value": "new rule text to add or replace with",
|
||||
"rationale": "why this change would help",
|
||||
"confidence": "high" | "medium" | "low"
|
||||
}}
|
||||
]
|
||||
}}
|
||||
|
||||
If there's insufficient data to make recommendations, return an empty suggestions array with an appropriate performance_summary."""
|
||||
|
||||
|
||||
class IntrospectionLLMResponse(IntrospectionReport):
|
||||
"""Schema for LLM response, excluding fields we'll fill in ourselves."""
|
||||
|
||||
# Override these to make them optional since LLM won't provide them
|
||||
workspace_name: str = ""
|
||||
generated_at: datetime.datetime = datetime.datetime.now(datetime.timezone.utc)
|
||||
signals: IntrospectionSignals = IntrospectionSignals()
|
||||
|
||||
|
||||
async def run_introspection(
|
||||
db: AsyncSession,
|
||||
workspace_name: str,
|
||||
) -> IntrospectionReport | None:
|
||||
"""
|
||||
Run introspection analysis for a workspace and generate a report.
|
||||
|
||||
Args:
|
||||
db: Database session
|
||||
workspace_name: Name of the workspace to analyze
|
||||
|
||||
Returns:
|
||||
IntrospectionReport with analysis and suggestions, or None if insufficient data
|
||||
"""
|
||||
logger.info(f"Starting introspection for workspace {workspace_name}")
|
||||
|
||||
# Gather signals
|
||||
signals = await gather_introspection_context(db, workspace_name)
|
||||
|
||||
# Check for minimum data
|
||||
if signals.total_dialectic_queries == 0 and signals.total_observations == 0:
|
||||
logger.info(
|
||||
f"Insufficient data for introspection in workspace {workspace_name}"
|
||||
)
|
||||
report = IntrospectionReport(
|
||||
workspace_name=workspace_name,
|
||||
generated_at=datetime.datetime.now(datetime.timezone.utc),
|
||||
performance_summary="Insufficient data for analysis. No dialectic queries or observations found.",
|
||||
identified_issues=[],
|
||||
suggestions=[],
|
||||
signals=signals,
|
||||
)
|
||||
await store_introspection_report(db, workspace_name, report)
|
||||
return report
|
||||
|
||||
# Build prompt and call LLM
|
||||
prompt = build_introspection_prompt(signals)
|
||||
|
||||
try:
|
||||
llm_response = await honcho_llm_call(
|
||||
llm_settings=settings.DREAM,
|
||||
prompt=prompt,
|
||||
max_tokens=4096,
|
||||
track_name="introspection",
|
||||
json_mode=True,
|
||||
temperature=0.3,
|
||||
)
|
||||
|
||||
# Parse the response
|
||||
response_text = (
|
||||
llm_response.content
|
||||
if hasattr(llm_response, "content")
|
||||
else str(llm_response)
|
||||
)
|
||||
|
||||
# Parse JSON and validate
|
||||
try:
|
||||
response_data: dict[str, object] = json.loads(response_text)
|
||||
except json.JSONDecodeError as e:
|
||||
logger.error(f"Failed to parse LLM response as JSON: {e}")
|
||||
response_data = {
|
||||
"performance_summary": f"Error parsing LLM response: {str(e)[:100]}",
|
||||
"identified_issues": [],
|
||||
"suggestions": [],
|
||||
}
|
||||
|
||||
# Build suggestions list
|
||||
suggestions: list[IntrospectionSuggestion] = []
|
||||
raw_suggestions = response_data.get("suggestions", [])
|
||||
if isinstance(raw_suggestions, list):
|
||||
for raw_suggestion in raw_suggestions:
|
||||
if isinstance(raw_suggestion, dict):
|
||||
try:
|
||||
# Validate target field
|
||||
target_raw = raw_suggestion.get("target", "deriver_rules")
|
||||
if target_raw not in ("deriver_rules", "dialectic_rules"):
|
||||
target_raw = "deriver_rules"
|
||||
|
||||
# Validate confidence field
|
||||
confidence_raw = raw_suggestion.get("confidence", "low")
|
||||
if confidence_raw not in ("high", "medium", "low"):
|
||||
confidence_raw = "low"
|
||||
|
||||
suggestions.append(
|
||||
IntrospectionSuggestion(
|
||||
target=target_raw, # type: ignore[arg-type]
|
||||
current_value=str(raw_suggestion.get("current_value", "")),
|
||||
suggested_value=str(raw_suggestion.get("suggested_value", "")),
|
||||
rationale=str(raw_suggestion.get("rationale", "")),
|
||||
confidence=confidence_raw, # type: ignore[arg-type]
|
||||
)
|
||||
)
|
||||
except Exception as e:
|
||||
logger.warning(f"Failed to parse suggestion: {e}")
|
||||
|
||||
# Extract fields with type coercion
|
||||
performance_summary = response_data.get("performance_summary", "Analysis completed.")
|
||||
if not isinstance(performance_summary, str):
|
||||
performance_summary = "Analysis completed."
|
||||
|
||||
identified_issues_raw = response_data.get("identified_issues", [])
|
||||
identified_issues: list[str] = []
|
||||
if isinstance(identified_issues_raw, list):
|
||||
for issue in identified_issues_raw:
|
||||
identified_issues.append(str(issue))
|
||||
|
||||
report = IntrospectionReport(
|
||||
workspace_name=workspace_name,
|
||||
generated_at=datetime.datetime.now(datetime.timezone.utc),
|
||||
performance_summary=performance_summary,
|
||||
identified_issues=identified_issues,
|
||||
suggestions=suggestions,
|
||||
signals=signals,
|
||||
)
|
||||
|
||||
except Exception as e:
|
||||
logger.error(f"LLM call failed during introspection: {e}")
|
||||
report = IntrospectionReport(
|
||||
workspace_name=workspace_name,
|
||||
generated_at=datetime.datetime.now(datetime.timezone.utc),
|
||||
performance_summary=f"Error during analysis: {str(e)[:200]}",
|
||||
identified_issues=[],
|
||||
suggestions=[],
|
||||
signals=signals,
|
||||
)
|
||||
|
||||
# Store the report
|
||||
await store_introspection_report(db, workspace_name, report)
|
||||
|
||||
logger.info(
|
||||
f"Introspection completed for {workspace_name}: {len(report.suggestions)} suggestions"
|
||||
)
|
||||
return report
|
||||
|
||||
|
||||
async def store_introspection_report(
|
||||
db: AsyncSession,
|
||||
workspace_name: str,
|
||||
report: IntrospectionReport,
|
||||
) -> None:
|
||||
"""
|
||||
Store an introspection report as a document in a reserved collection.
|
||||
|
||||
Reports are stored with:
|
||||
- observer: _system
|
||||
- observed: _introspection
|
||||
- content: JSON-serialized report
|
||||
- embedding: None (not for semantic search)
|
||||
|
||||
Args:
|
||||
db: Database session
|
||||
workspace_name: Name of the workspace
|
||||
report: The introspection report to store
|
||||
"""
|
||||
from src import schemas
|
||||
|
||||
try:
|
||||
# Ensure system peers exist
|
||||
await crud.get_or_create_peers(
|
||||
db,
|
||||
workspace_name,
|
||||
[
|
||||
schemas.PeerCreate(name=SYSTEM_OBSERVER),
|
||||
schemas.PeerCreate(name=INTROSPECTION_OBSERVED),
|
||||
],
|
||||
)
|
||||
|
||||
# Ensure collection exists
|
||||
await crud.get_or_create_collection(
|
||||
db,
|
||||
workspace_name,
|
||||
observer=SYSTEM_OBSERVER,
|
||||
observed=INTROSPECTION_OBSERVED,
|
||||
)
|
||||
|
||||
# Serialize report to JSON
|
||||
report_json = report.model_dump_json()
|
||||
|
||||
# Create document (without embedding - not for semantic search)
|
||||
doc = models.Document(
|
||||
workspace_name=workspace_name,
|
||||
observer=SYSTEM_OBSERVER,
|
||||
observed=INTROSPECTION_OBSERVED,
|
||||
content=report_json,
|
||||
level="explicit",
|
||||
times_derived=1,
|
||||
internal_metadata={
|
||||
"report_type": "introspection",
|
||||
"generated_at": report.generated_at.isoformat(),
|
||||
},
|
||||
session_name=None,
|
||||
embedding=None,
|
||||
sync_state="synced", # No need to sync - no embedding
|
||||
)
|
||||
|
||||
db.add(doc)
|
||||
await db.commit()
|
||||
|
||||
logger.debug(
|
||||
f"Stored introspection report for {workspace_name}, doc_id={doc.id}"
|
||||
)
|
||||
|
||||
except Exception as e:
|
||||
logger.error(f"Failed to store introspection report: {e}")
|
||||
await db.rollback()
|
||||
# Don't re-raise - storing the report is secondary to generating it
|
||||
|
|
@ -360,6 +360,17 @@ DREAM: {payload.dream_type} documents for {workspace_name}/{payload.observer}/{p
|
|||
+ f"duration={result.total_duration_ms:.0f}ms"
|
||||
)
|
||||
|
||||
case DreamType.INTROSPECTION:
|
||||
from src.dreamer.introspection import run_introspection
|
||||
|
||||
async with tracked_db("introspection") as db:
|
||||
result = await run_introspection(db, workspace_name)
|
||||
|
||||
if result is not None:
|
||||
logger.info(
|
||||
f"Introspection completed: {len(result.suggestions)} suggestions"
|
||||
)
|
||||
|
||||
except Exception as e:
|
||||
logger.error(
|
||||
f"Error processing dream task {payload.dream_type} for {payload.observer}/{payload.observed}: {str(e)}",
|
||||
|
|
|
|||
|
|
@ -1,7 +1,7 @@
|
|||
import datetime
|
||||
import ipaddress
|
||||
from enum import Enum
|
||||
from typing import Annotated, Any, Self, cast
|
||||
from typing import Annotated, Any, Literal, Self, cast
|
||||
from urllib.parse import urlparse
|
||||
|
||||
import tiktoken
|
||||
|
|
@ -25,6 +25,7 @@ class DreamType(str, Enum):
|
|||
"""Types of dreams that can be triggered."""
|
||||
|
||||
OMNI = "omni"
|
||||
INTROSPECTION = "introspection"
|
||||
|
||||
|
||||
class WorkspaceAgentConfig(BaseModel):
|
||||
|
|
@ -44,6 +45,44 @@ class WorkspaceAgentConfig(BaseModel):
|
|||
)
|
||||
|
||||
|
||||
class IntrospectionSignals(BaseModel):
|
||||
"""Signals gathered for introspection analysis."""
|
||||
|
||||
total_dialectic_queries: int = 0
|
||||
avg_dialectic_duration_ms: float = 0.0
|
||||
abstention_count: int = 0
|
||||
abstention_rate: float = 0.0
|
||||
recent_queries: list[str] = Field(default_factory=list)
|
||||
total_observations: int = 0
|
||||
observations_by_level: dict[str, int] = Field(default_factory=dict)
|
||||
contradiction_count: int = 0
|
||||
total_peers: int = 0
|
||||
total_sessions: int = 0
|
||||
current_deriver_rules: str = ""
|
||||
current_dialectic_rules: str = ""
|
||||
|
||||
|
||||
class IntrospectionSuggestion(BaseModel):
|
||||
"""A single configuration suggestion."""
|
||||
|
||||
target: Literal["deriver_rules", "dialectic_rules"]
|
||||
current_value: str
|
||||
suggested_value: str
|
||||
rationale: str
|
||||
confidence: Literal["high", "medium", "low"]
|
||||
|
||||
|
||||
class IntrospectionReport(BaseModel):
|
||||
"""Complete introspection report for a workspace."""
|
||||
|
||||
workspace_name: str
|
||||
generated_at: datetime.datetime
|
||||
performance_summary: str
|
||||
identified_issues: list[str] = Field(default_factory=list)
|
||||
suggestions: list[IntrospectionSuggestion] = Field(default_factory=list)
|
||||
signals: IntrospectionSignals
|
||||
|
||||
|
||||
class ReconcilerType(str, Enum):
|
||||
"""Types of reconciler tasks that can be performed."""
|
||||
|
||||
|
|
|
|||
|
|
@ -0,0 +1,423 @@
|
|||
"""Tests for the introspection module."""
|
||||
|
||||
import datetime
|
||||
import json
|
||||
from unittest.mock import AsyncMock, MagicMock, patch
|
||||
|
||||
import pytest
|
||||
from nanoid import generate as generate_nanoid
|
||||
from sqlalchemy.ext.asyncio import AsyncSession
|
||||
|
||||
from src import crud, models
|
||||
from src.dreamer.introspection import (
|
||||
INTROSPECTION_OBSERVED,
|
||||
SYSTEM_OBSERVER,
|
||||
build_introspection_prompt,
|
||||
gather_introspection_context,
|
||||
run_introspection,
|
||||
store_introspection_report,
|
||||
)
|
||||
from src.schemas import (
|
||||
DialecticTraceCreate,
|
||||
IntrospectionReport,
|
||||
IntrospectionSignals,
|
||||
IntrospectionSuggestion,
|
||||
)
|
||||
|
||||
|
||||
class TestGatherIntrospectionContext:
|
||||
"""Tests for gathering introspection signals."""
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_gather_introspection_context_empty_workspace(
|
||||
self,
|
||||
db_session: AsyncSession,
|
||||
sample_data: tuple[models.Workspace, models.Peer],
|
||||
):
|
||||
"""Test gathering signals from an empty workspace returns zero values."""
|
||||
workspace, _ = sample_data
|
||||
|
||||
signals = await gather_introspection_context(db_session, workspace.name)
|
||||
|
||||
assert signals.total_dialectic_queries == 0
|
||||
assert signals.avg_dialectic_duration_ms == 0.0
|
||||
assert signals.abstention_count == 0
|
||||
assert signals.abstention_rate == 0.0
|
||||
assert signals.recent_queries == []
|
||||
assert signals.total_observations == 0
|
||||
assert signals.observations_by_level == {}
|
||||
assert signals.contradiction_count == 0
|
||||
# Peer count may be 1 (the sample peer), but should not include system peers
|
||||
assert signals.total_sessions == 0
|
||||
assert signals.current_deriver_rules == ""
|
||||
assert signals.current_dialectic_rules == ""
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_gather_introspection_context_with_data(
|
||||
self,
|
||||
db_session: AsyncSession,
|
||||
sample_data: tuple[models.Workspace, models.Peer],
|
||||
):
|
||||
"""Test gathering signals from a workspace with data."""
|
||||
workspace, peer = sample_data
|
||||
|
||||
# Create some dialectic traces
|
||||
for i in range(5):
|
||||
trace_data = DialecticTraceCreate(
|
||||
workspace_name=workspace.name,
|
||||
observer=peer.name,
|
||||
observed=peer.name,
|
||||
query=f"What does the user like? Query {i}",
|
||||
response="The user likes coffee." if i < 3 else "I don't have information.",
|
||||
reasoning_level="low",
|
||||
total_duration_ms=100.0 * (i + 1),
|
||||
input_tokens=100,
|
||||
output_tokens=50,
|
||||
)
|
||||
await crud.create_dialectic_trace(db_session, trace_data)
|
||||
|
||||
# Create a session
|
||||
session = models.Session(name=generate_nanoid(), workspace_name=workspace.name)
|
||||
db_session.add(session)
|
||||
await db_session.flush()
|
||||
|
||||
signals = await gather_introspection_context(db_session, workspace.name)
|
||||
|
||||
assert signals.total_dialectic_queries == 5
|
||||
assert signals.avg_dialectic_duration_ms == 300.0 # (100+200+300+400+500)/5
|
||||
assert signals.abstention_count == 2 # Last 2 responses are abstentions
|
||||
assert signals.abstention_rate == 0.4
|
||||
assert len(signals.recent_queries) == 5
|
||||
assert "What does the user like?" in signals.recent_queries[0]
|
||||
assert signals.total_sessions == 1
|
||||
|
||||
|
||||
class TestBuildIntrospectionPrompt:
|
||||
"""Tests for building the introspection prompt."""
|
||||
|
||||
def test_build_introspection_prompt_empty_signals(self):
|
||||
"""Test prompt building with empty signals."""
|
||||
signals = IntrospectionSignals()
|
||||
|
||||
prompt = build_introspection_prompt(signals)
|
||||
|
||||
assert "Total queries: 0" in prompt
|
||||
assert "Abstention rate: 0.0%" in prompt
|
||||
assert "(No recent queries)" in prompt
|
||||
assert "(No observations)" in prompt
|
||||
assert "(empty - using defaults)" in prompt
|
||||
|
||||
def test_build_introspection_prompt_with_data(self):
|
||||
"""Test prompt building with populated signals."""
|
||||
signals = IntrospectionSignals(
|
||||
total_dialectic_queries=100,
|
||||
avg_dialectic_duration_ms=250.5,
|
||||
abstention_count=30,
|
||||
abstention_rate=0.3,
|
||||
recent_queries=["What is the user's name?", "What do they prefer?"],
|
||||
total_observations=500,
|
||||
observations_by_level={"explicit": 300, "deductive": 150, "inductive": 50},
|
||||
contradiction_count=5,
|
||||
total_peers=10,
|
||||
total_sessions=25,
|
||||
current_deriver_rules="Focus on preferences",
|
||||
current_dialectic_rules="Be concise",
|
||||
)
|
||||
|
||||
prompt = build_introspection_prompt(signals)
|
||||
|
||||
assert "Total queries: 100" in prompt
|
||||
assert "250.5ms" in prompt
|
||||
assert "Abstention rate: 30.0%" in prompt
|
||||
assert "What is the user's name?" in prompt
|
||||
assert "explicit: 300" in prompt
|
||||
assert "deductive: 150" in prompt
|
||||
assert "Contradictions detected: 5" in prompt
|
||||
assert "Total peers: 10" in prompt
|
||||
assert "Total sessions: 25" in prompt
|
||||
assert "Focus on preferences" in prompt
|
||||
assert "Be concise" in prompt
|
||||
|
||||
|
||||
class TestRunIntrospection:
|
||||
"""Tests for the main introspection runner."""
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_run_introspection_insufficient_data(
|
||||
self,
|
||||
db_session: AsyncSession,
|
||||
sample_data: tuple[models.Workspace, models.Peer],
|
||||
):
|
||||
"""Test introspection with insufficient data returns appropriate report."""
|
||||
workspace, _ = sample_data
|
||||
|
||||
with patch(
|
||||
"src.dreamer.introspection.store_introspection_report",
|
||||
new_callable=AsyncMock,
|
||||
) as mock_store:
|
||||
report = await run_introspection(db_session, workspace.name)
|
||||
|
||||
assert report is not None
|
||||
assert report.workspace_name == workspace.name
|
||||
assert "Insufficient data" in report.performance_summary
|
||||
assert report.suggestions == []
|
||||
assert report.identified_issues == []
|
||||
mock_store.assert_called_once()
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_run_introspection_with_data(
|
||||
self,
|
||||
db_session: AsyncSession,
|
||||
sample_data: tuple[models.Workspace, models.Peer],
|
||||
):
|
||||
"""Test introspection with data calls LLM and parses response."""
|
||||
workspace, peer = sample_data
|
||||
|
||||
# Create some dialectic traces
|
||||
for i in range(3):
|
||||
trace_data = DialecticTraceCreate(
|
||||
workspace_name=workspace.name,
|
||||
observer=peer.name,
|
||||
observed=peer.name,
|
||||
query=f"Query {i}",
|
||||
response="Response",
|
||||
reasoning_level="low",
|
||||
total_duration_ms=100.0,
|
||||
input_tokens=100,
|
||||
output_tokens=50,
|
||||
)
|
||||
await crud.create_dialectic_trace(db_session, trace_data)
|
||||
|
||||
# Mock the LLM call
|
||||
mock_llm_response = MagicMock()
|
||||
mock_llm_response.content = json.dumps({
|
||||
"performance_summary": "The workspace is performing well.",
|
||||
"identified_issues": ["High abstention rate"],
|
||||
"suggestions": [
|
||||
{
|
||||
"target": "deriver_rules",
|
||||
"current_value": "",
|
||||
"suggested_value": "Focus on capturing user preferences",
|
||||
"rationale": "Too many queries about preferences are being missed",
|
||||
"confidence": "medium",
|
||||
}
|
||||
],
|
||||
})
|
||||
|
||||
with (
|
||||
patch(
|
||||
"src.dreamer.introspection.honcho_llm_call",
|
||||
new_callable=AsyncMock,
|
||||
return_value=mock_llm_response,
|
||||
),
|
||||
patch(
|
||||
"src.dreamer.introspection.store_introspection_report",
|
||||
new_callable=AsyncMock,
|
||||
) as mock_store,
|
||||
):
|
||||
report = await run_introspection(db_session, workspace.name)
|
||||
|
||||
assert report is not None
|
||||
assert report.workspace_name == workspace.name
|
||||
assert report.performance_summary == "The workspace is performing well."
|
||||
assert "High abstention rate" in report.identified_issues
|
||||
assert len(report.suggestions) == 1
|
||||
assert report.suggestions[0].target == "deriver_rules"
|
||||
assert report.suggestions[0].confidence == "medium"
|
||||
mock_store.assert_called_once()
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_run_introspection_llm_failure(
|
||||
self,
|
||||
db_session: AsyncSession,
|
||||
sample_data: tuple[models.Workspace, models.Peer],
|
||||
):
|
||||
"""Test introspection handles LLM failures gracefully."""
|
||||
workspace, peer = sample_data
|
||||
|
||||
# Create some dialectic traces
|
||||
trace_data = DialecticTraceCreate(
|
||||
workspace_name=workspace.name,
|
||||
observer=peer.name,
|
||||
observed=peer.name,
|
||||
query="Query",
|
||||
response="Response",
|
||||
reasoning_level="low",
|
||||
total_duration_ms=100.0,
|
||||
input_tokens=100,
|
||||
output_tokens=50,
|
||||
)
|
||||
await crud.create_dialectic_trace(db_session, trace_data)
|
||||
|
||||
with (
|
||||
patch(
|
||||
"src.dreamer.introspection.honcho_llm_call",
|
||||
new_callable=AsyncMock,
|
||||
side_effect=Exception("LLM API error"),
|
||||
),
|
||||
patch(
|
||||
"src.dreamer.introspection.store_introspection_report",
|
||||
new_callable=AsyncMock,
|
||||
) as mock_store,
|
||||
):
|
||||
report = await run_introspection(db_session, workspace.name)
|
||||
|
||||
assert report is not None
|
||||
assert "Error during analysis" in report.performance_summary
|
||||
assert report.suggestions == []
|
||||
mock_store.assert_called_once()
|
||||
|
||||
|
||||
class TestStoreIntrospectionReport:
|
||||
"""Tests for storing introspection reports."""
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_store_introspection_report_creates_document(
|
||||
self,
|
||||
db_session: AsyncSession,
|
||||
sample_data: tuple[models.Workspace, models.Peer],
|
||||
):
|
||||
"""Test that storing a report creates a document in the reserved collection."""
|
||||
from src.crud.collection import get_or_create_collection as real_get_or_create_collection
|
||||
|
||||
workspace, _ = sample_data
|
||||
# Capture workspace name before any potential session issues
|
||||
workspace_name = workspace.name
|
||||
|
||||
# Create system peers and collection directly (bypassing mock)
|
||||
system_peer = models.Peer(name=SYSTEM_OBSERVER, workspace_name=workspace_name)
|
||||
introspection_peer = models.Peer(
|
||||
name=INTROSPECTION_OBSERVED, workspace_name=workspace_name
|
||||
)
|
||||
db_session.add_all([system_peer, introspection_peer])
|
||||
await db_session.flush()
|
||||
|
||||
# Create the collection
|
||||
collection = models.Collection(
|
||||
workspace_name=workspace_name,
|
||||
observer=SYSTEM_OBSERVER,
|
||||
observed=INTROSPECTION_OBSERVED,
|
||||
)
|
||||
db_session.add(collection)
|
||||
await db_session.flush()
|
||||
|
||||
report = IntrospectionReport(
|
||||
workspace_name=workspace_name,
|
||||
generated_at=datetime.datetime.now(datetime.timezone.utc),
|
||||
performance_summary="Test summary",
|
||||
identified_issues=["Issue 1"],
|
||||
suggestions=[
|
||||
IntrospectionSuggestion(
|
||||
target="deriver_rules",
|
||||
current_value="",
|
||||
suggested_value="New rule",
|
||||
rationale="Test rationale",
|
||||
confidence="high",
|
||||
)
|
||||
],
|
||||
signals=IntrospectionSignals(total_dialectic_queries=10),
|
||||
)
|
||||
|
||||
await store_introspection_report(db_session, workspace_name, report)
|
||||
|
||||
# Verify document was created
|
||||
from sqlalchemy import select
|
||||
|
||||
stmt = (
|
||||
select(models.Document)
|
||||
.where(models.Document.workspace_name == workspace_name)
|
||||
.where(models.Document.observer == SYSTEM_OBSERVER)
|
||||
.where(models.Document.observed == INTROSPECTION_OBSERVED)
|
||||
)
|
||||
result = await db_session.execute(stmt)
|
||||
doc = result.scalar_one_or_none()
|
||||
|
||||
assert doc is not None
|
||||
assert doc.level == "explicit"
|
||||
assert doc.embedding is None # No embedding for reports
|
||||
assert doc.sync_state == "synced"
|
||||
|
||||
# Verify content is valid JSON
|
||||
content = json.loads(doc.content)
|
||||
assert content["workspace_name"] == workspace_name
|
||||
assert content["performance_summary"] == "Test summary"
|
||||
assert len(content["suggestions"]) == 1
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_store_introspection_report_creates_system_peers(
|
||||
self,
|
||||
db_session: AsyncSession,
|
||||
sample_data: tuple[models.Workspace, models.Peer],
|
||||
):
|
||||
"""Test that storing a report creates the _system and _introspection peers."""
|
||||
workspace, _ = sample_data
|
||||
# Capture workspace name before any potential session issues
|
||||
workspace_name = workspace.name
|
||||
|
||||
report = IntrospectionReport(
|
||||
workspace_name=workspace_name,
|
||||
generated_at=datetime.datetime.now(datetime.timezone.utc),
|
||||
performance_summary="Test",
|
||||
signals=IntrospectionSignals(),
|
||||
)
|
||||
|
||||
await store_introspection_report(db_session, workspace_name, report)
|
||||
|
||||
# Verify system peers were created
|
||||
from sqlalchemy import select
|
||||
|
||||
stmt = (
|
||||
select(models.Peer)
|
||||
.where(models.Peer.workspace_name == workspace_name)
|
||||
.where(models.Peer.name.in_([SYSTEM_OBSERVER, INTROSPECTION_OBSERVED]))
|
||||
)
|
||||
result = await db_session.execute(stmt)
|
||||
peers = list(result.scalars().all())
|
||||
|
||||
assert len(peers) == 2
|
||||
peer_names = {p.name for p in peers}
|
||||
assert SYSTEM_OBSERVER in peer_names
|
||||
assert INTROSPECTION_OBSERVED in peer_names
|
||||
|
||||
|
||||
class TestIntrospectionDreamDispatch:
|
||||
"""Tests for introspection via the dream dispatch mechanism."""
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_introspection_dream_type_dispatch(
|
||||
self,
|
||||
db_session: AsyncSession,
|
||||
sample_data: tuple[models.Workspace, models.Peer],
|
||||
):
|
||||
"""Test that INTROSPECTION dream type is properly dispatched."""
|
||||
from src.dreamer.orchestrator import process_dream
|
||||
from src.schemas import DreamType
|
||||
from src.utils.queue_payload import DreamPayload
|
||||
|
||||
workspace, peer = sample_data
|
||||
|
||||
payload = DreamPayload(
|
||||
dream_type=DreamType.INTROSPECTION,
|
||||
observer=peer.name,
|
||||
observed=peer.name,
|
||||
session_name=None,
|
||||
)
|
||||
|
||||
with patch(
|
||||
"src.dreamer.introspection.run_introspection",
|
||||
new_callable=AsyncMock,
|
||||
) as mock_run:
|
||||
mock_run.return_value = IntrospectionReport(
|
||||
workspace_name=workspace.name,
|
||||
generated_at=datetime.datetime.now(datetime.timezone.utc),
|
||||
performance_summary="Test",
|
||||
suggestions=[],
|
||||
signals=IntrospectionSignals(),
|
||||
)
|
||||
|
||||
await process_dream(payload, workspace.name)
|
||||
|
||||
mock_run.assert_called_once()
|
||||
# Verify it was called with the workspace name
|
||||
call_args = mock_run.call_args
|
||||
assert call_args[0][1] == workspace.name
|
||||
Loading…
Reference in New Issue