# OrchestKit Langfuse Traces - Real Implementation

This document shows how OrchestKit uses Langfuse for end-to-end LLM observability across its 8-agent LangGraph workflow.

## Overview

**OrchestKit Analysis Pipeline:**
- 8 specialized agents (Tech Comparator, Security Auditor, Implementation Planner, etc.)
- LangGraph supervisor pattern for orchestration
- Langfuse traces for cost tracking, performance monitoring, and debugging

**Migration**: LangSmith → Langfuse (December 2025)
- Self-hosted, open-source, free
- Better prompt management
- Native cost tracking
- Session-based grouping

## Trace Architecture

### Analysis Session Structure

```
content_analysis (session_id: analysis_550e8400)
├── fetch_content (0.3s)
│   └── metadata: {url, content_size_bytes: 45823}
├── generate_embedding (0.8s, $0.0002)
│   └── model: voyage-code-2
│   └── tokens: 11,456 input
└── supervisor_workflow (12.5s, $0.145)
    ├── supervisor_route_1 (0.1s)
    │   └── next_agent: tech_comparator
    ├── tech_comparator (2.1s, $0.018)
    │   ├── analyze_technologies (1.8s, $0.015)
    │   │   └── model: claude-sonnet-4-6
    │   │   └── tokens: 1,500 input, 1,000 output
    │   └── compress_findings (0.2s, $0.003)
    │       └── model: claude-sonnet-4-6
    │       └── tokens: 800 input, 400 output
    ├── supervisor_route_2 (0.1s)
    │   └── next_agent: security_auditor
    ├── security_auditor (2.3s, $0.021)
    │   └── ... (similar structure)
    ├── ... (6 more agents)
    └── quality_gate (1.2s, $0.012)
        ├── g_eval_completeness (0.4s, $0.004)
        ├── g_eval_accuracy (0.4s, $0.004)
        ├── g_eval_coherence (0.2s, $0.002)
        └── g_eval_depth (0.2s, $0.002)
```

**Session Metrics:**
- Total duration: 15.4s
- Total cost: $0.147
- Agents executed: 8
- Quality scores: completeness=0.85, accuracy=0.92, coherence=0.88, depth=0.78

## Implementation Examples

### 1. Workflow-Level Tracing

**File:** `backend/app/domains/analysis/workflows/content_analysis.py`

```python
from langfuse import observe, get_client
from app.shared.services.langfuse.client import langfuse_client

@observe(name="content_analysis_workflow")
async def run_content_analysis(analysis_id: str, url: str) -> AnalysisResult:
    """Analyze content with 8-agent supervisor workflow."""

    # Set session-level metadata
    get_client().update_current_trace(
        name="content_analysis",
        session_id=f"analysis_{analysis_id}",
        user_id="system",
        metadata={
            "analysis_id": analysis_id,
            "url": url,
            "workflow_type": "8-agent-supervisor",
            "version": "1.0.0"
        },
        tags=["production", "orchestkit", "langgraph"]
    )

    # Step 1: Fetch content (nested span)
    content = await fetch_content(url)  # @observe decorated

    # Step 2: Generate embedding (nested span with cost tracking)
    embedding = await generate_embedding(content)  # @observe decorated

    # Step 3: Run supervisor workflow (8 agents in parallel/sequential)
    findings = await run_supervisor_workflow(content)

    # Track total cost
    total_cost = sum(f.cost_usd for f in findings)
    get_client().update_current_observation(
        metadata={
            "total_agents": len(findings),
            "total_cost_usd": total_cost,
            "total_tokens": sum(f.token_count for f in findings)
        }
    )

    return AnalysisResult(findings=findings, total_cost=total_cost)
```

### 2. Agent-Level Tracing

**File:** `backend/app/domains/analysis/workflows/nodes/agent_node.py`

```python
@observe(name="agent_execution")
async def execute_agent(
    agent_type: str,
    content: str,
    state: AnalysisState
) -> Finding:
    """Execute single agent with Langfuse tracing."""

    # Set agent-specific context
    get_client().update_current_observation(
        name=f"agent_{agent_type}",
        metadata={
            "agent_type": agent_type,
            "content_length": len(content),
            "correlation_id": state["correlation_id"]
        }
    )

    # Call LLM with automatic cost tracking
    response = await call_llm_with_tracing(
        agent_type=agent_type,
        content=content,
        state=state
    )

    # Score the response
    quality_scores = await score_agent_output(agent_type, response)

    # Add scores to trace
    for criterion, score in quality_scores.items():
        get_client().score(
            name=f"{agent_type}_{criterion}",
            value=score,
            data_type="NUMERIC"
        )

    return response
```

### 3. LLM Call Tracing with Cost Tracking

**File:** `backend/app/shared/services/llm/anthropic_client.py`

```python
from langfuse import observe, get_client

@observe(name="llm_call")
async def call_anthropic(
    messages: list[dict],
    model: str = "claude-sonnet-4-6",
    **kwargs
) -> str:
    """Call Anthropic with automatic Langfuse cost tracking."""

    # Log input (truncated for large prompts)
    get_client().update_current_observation(
        input=str(messages)[:2000],
        model=model,
        metadata={
            "temperature": kwargs.get("temperature", 1.0),
            "max_tokens": kwargs.get("max_tokens", 4096)
        }
    )

    # Call Anthropic API
    response = await anthropic_client.messages.create(
        model=model,
        messages=messages,
        **kwargs
    )

    # Extract token usage
    input_tokens = response.usage.input_tokens
    output_tokens = response.usage.output_tokens

    # Cost calculation (Claude Sonnet 4.5 pricing)
    input_cost = (input_tokens / 1_000_000) * 3.00   # $3/MTok
    output_cost = (output_tokens / 1_000_000) * 15.00  # $15/MTok
    total_cost = input_cost + output_cost

    # Log output and costs to Langfuse
    get_client().update_current_observation(
        output=response.content[0].text[:2000],
        usage={
            "input": input_tokens,
            "output": output_tokens,
            "unit": "TOKENS"
        },
        metadata={
            "cost_usd": total_cost,
            "input_cost_usd": input_cost,
            "output_cost_usd": output_cost,
            "prompt_caching_enabled": kwargs.get("cache_control") is not None
        }
    )

    logger.info("llm_call_completed",
        model=model,
        input_tokens=input_tokens,
        output_tokens=output_tokens,
        cost_usd=total_cost,
        cache_enabled=kwargs.get("cache_control") is not None
    )

    return response.content[0].text
```

### 4. Quality Gate Evaluation Tracing

**File:** `backend/app/workflows/nodes/quality_gate_node.py`

```python
@observe(name="quality_gate")
async def quality_gate_node(state: AnalysisState) -> AnalysisState:
    """Evaluate aggregated findings with G-Eval scoring."""

    get_client().update_current_observation(
        metadata={
            "findings_count": len(state["findings"]),
            "analysis_id": state["analysis_id"]
        }
    )

    # Run G-Eval for 4 criteria in parallel
    criteria = ["completeness", "accuracy", "coherence", "depth"]

    scores = await asyncio.gather(*[
        evaluate_criterion(criterion, state["findings"])
        for criterion in criteria
    ])

    # Log individual criterion scores
    score_dict = {}
    for criterion, score in zip(criteria, scores):
        score_dict[criterion] = score
        get_client().score(
            name=f"quality_{criterion}",
            value=score,
            comment=f"G-Eval score for {criterion} criterion"
        )

    # Overall quality score (weighted average)
    overall_quality = (
        score_dict["completeness"] * 0.3 +
        score_dict["accuracy"] * 0.3 +
        score_dict["coherence"] * 0.2 +
        score_dict["depth"] * 0.2
    )

    get_client().score(
        name="quality_overall",
        value=overall_quality,
        comment="Weighted average of all criteria"
    )

    state["quality_scores"] = score_dict
    state["overall_quality"] = overall_quality

    return state
```

## Real Metrics from Production

### Cost Breakdown by Agent

**Langfuse Query (Last 30 Days):**
```sql
SELECT
    metadata->>'agent_type' as agent,
    COUNT(*) as executions,
    AVG(calculated_total_cost) as avg_cost,
    SUM(calculated_total_cost) as total_cost,
    AVG(input_tokens) as avg_input_tokens,
    AVG(output_tokens) as avg_output_tokens
FROM traces
WHERE metadata->>'agent_type' IS NOT NULL
    AND timestamp > NOW() - INTERVAL '30 days'
GROUP BY agent
ORDER BY total_cost DESC;
```

**Results:**
| Agent | Executions | Avg Cost | Total Cost | Avg Input | Avg Output |
|-------|------------|----------|------------|-----------|------------|
| security_auditor | 145 | $0.021 | $3.05 | 1,800 | 1,200 |
| implementation_planner | 145 | $0.019 | $2.76 | 1,600 | 1,100 |
| tech_comparator | 145 | $0.018 | $2.61 | 1,500 | 1,000 |
| performance_analyzer | 145 | $0.017 | $2.47 | 1,400 | 950 |
| quality_gate | 145 | $0.012 | $1.74 | 1,000 | 600 |
| architecture_reviewer | 145 | $0.015 | $2.18 | 1,300 | 900 |
| testing_strategist | 145 | $0.014 | $2.03 | 1,200 | 850 |
| documentation_expert | 145 | $0.013 | $1.89 | 1,100 | 800 |

**Insights:**
- Security Auditor is most expensive (detailed vulnerability analysis)
- Quality Gate is cheapest (focused evaluation)
- Total monthly cost: $18.73 (145 analyses)
- Average per analysis: $0.129

### Cache Hit Impact

**Before Caching (Dec 2024):**
- Monthly cost: $35,000 (projected annual: $420k)
- Average latency: 2.1s per LLM call

**After Multi-Level Caching (Jan 2025):**
- L1 (Prompt Cache): 90% hit rate → $31,500 saved (90% savings on cache hits)
- L2 (Semantic Cache): 75% hit rate on L1 misses → $2,625 saved (85% savings)
- Final monthly cost: $875
- **Total savings: 97.5%**
- Average latency: 5-10ms (semantic cache hit)

**Langfuse Cache Analytics:**
```sql
-- Cache hit rate by agent
SELECT
    metadata->>'agent_type' as agent,
    COUNT(*) FILTER (WHERE metadata->>'cache_hit' = 'true') as cache_hits,
    COUNT(*) as total_calls,
    ROUND(100.0 * COUNT(*) FILTER (WHERE metadata->>'cache_hit' = 'true') / COUNT(*), 2) as hit_rate_pct
FROM traces
WHERE metadata->>'agent_type' IS NOT NULL
GROUP BY agent
ORDER BY hit_rate_pct DESC;
```

**Results:**
| Agent | Cache Hits | Total Calls | Hit Rate |
|-------|------------|-------------|----------|
| tech_comparator | 133 | 145 | 91.7% |
| performance_analyzer | 128 | 145 | 88.3% |
| testing_strategist | 125 | 145 | 86.2% |
| security_auditor | 58 | 145 | 40.0% |

**Why security_auditor has low cache hit rate:**
- Unique vulnerabilities per codebase
- Security context is highly specific
- **Opportunity**: Implement vulnerability pattern caching

## Dashboard Queries

### Top 10 Most Expensive Analyses

```sql
SELECT
    name,
    session_id,
    calculated_total_cost as cost_usd,
    timestamp,
    metadata->>'url' as analyzed_url,
    metadata->>'total_agents' as agents_executed
FROM traces
WHERE name = 'content_analysis_workflow'
ORDER BY calculated_total_cost DESC
LIMIT 10;
```

### Quality Trend Over Time

```sql
SELECT
    DATE(timestamp) as date,
    AVG(value) FILTER (WHERE name = 'quality_completeness') as avg_completeness,
    AVG(value) FILTER (WHERE name = 'quality_accuracy') as avg_accuracy,
    AVG(value) FILTER (WHERE name = 'quality_coherence') as avg_coherence,
    AVG(value) FILTER (WHERE name = 'quality_depth') as avg_depth,
    AVG(value) FILTER (WHERE name = 'quality_overall') as avg_overall
FROM scores
WHERE timestamp > NOW() - INTERVAL '30 days'
GROUP BY DATE(timestamp)
ORDER BY date;
```

**Results (Last 30 Days):**
| Date | Completeness | Accuracy | Coherence | Depth | Overall |
|------|--------------|----------|-----------|-------|---------|
| 2025-01-15 | 0.83 | 0.91 | 0.87 | 0.76 | 0.84 |
| 2025-01-16 | 0.85 | 0.92 | 0.88 | 0.78 | 0.86 |
| 2025-01-17 | 0.84 | 0.90 | 0.86 | 0.75 | 0.84 |

**Trend:** Quality scores stable, depth improving (+4% since truncation fix)

### Slow Trace Detection

```sql
-- Find traces slower than 2 standard deviations
WITH stats AS (
    SELECT
        AVG(latency_seconds) as mean,
        STDDEV(latency_seconds) as stddev
    FROM traces
    WHERE name = 'content_analysis_workflow'
)
SELECT
    t.session_id,
    t.latency_seconds,
    t.metadata->>'url' as url,
    t.timestamp
FROM traces t, stats s
WHERE t.name = 'content_analysis_workflow'
    AND t.latency_seconds > (s.mean + 2 * s.stddev)
ORDER BY t.latency_seconds DESC
LIMIT 20;
```

## Best Practices from OrchestKit

1. **Always use @observe decorator** - Automatic parent-child span relationships
2. **Set session_id for multi-step workflows** - Group related traces together
3. **Tag production vs staging** - Filter by environment
4. **Add agent_type to metadata** - Enable cost/performance analysis by agent
5. **Log truncated inputs/outputs** - Keep traces small (2000 chars max)
6. **Score all quality metrics** - Enable quality trend monitoring
7. **Track cache_hit in metadata** - Measure caching effectiveness
8. **Use correlation_id across services** - Link to application logs

## References

- [Langfuse Self-Hosting Guide](https://langfuse.com/docs/deployment/self-host)
- [Python SDK Decorators](https://langfuse.com/docs/sdk/python/decorators)
- [Cost Tracking](https://langfuse.com/docs/model-usage-and-cost)
- [OrchestKit QUALITY_INITIATIVE_FIXES.md](../../../docs/QUALITY_INITIATIVE_FIXES.md)
