Complete Examples

This page provides complete, production-ready examples using the CAIP Agents SDK with full Langfuse observability. Each example includes all imports, setup code, and produces visible traces in Langfuse.

Prerequisites

Before running any example:

Set up your .env file with Langfuse credentials
Install the SDK (see Installation)
Set your CAIP agent ID: export CAIP_AGENT_ID=your-agent-id
Optional region routing: export CAIP_REGION=ROW (ROW default, CN supported)

.env (Required for all examples)
LANGFUSE_PUBLIC_KEY=pk-lf-your-key-here
LANGFUSE_SECRET_KEY=sk-lf-your-key-here
# ROW
LANGFUSE_HOST=https://langfuse.caip.bmw.cloud

# CN
# LANGFUSE_HOST=https://langfuse.caip.bmwchina.cloud
LANGFUSE_ENV=prod
CAIP_AGENT_ID=your-agent-id
CAIP_REGION=ROW  # Optional: ROW (default) or CN
# Optional: disable Langfuse observability
# CAIP_LANGFUSE_OBSERVABILITY=false

User ID Convention

All examples use QX11111 as the user ID for consistent trace attribution. In production, replace this with your actual user identifier.

Example 1: Simple Agent Tracing

What it demonstrates: Basic CAIP agent with @observe decorator, user context propagation, and chain observation to consolidate all operations into a single trace.

Key Concepts:

CAIPAgentsClient initialization
@observe decorator for function tracing
propagate_attributes for user/session context
as_type="chain" to consolidate operations
score_current_trace for quality metrics

example_simple_tracing.py
"""
Example 1: Simple Agent Tracing with CAIP Agents SDK
=====================================================
Basic observability setup with a CAIP agent and trace context propagation.
"""

import asyncio
import os
import time

from caip_agents_sdk import (
    CAIPAgentsClient,
    observe,
    propagate_attributes,
    flush_traces,
    enable_logging,
)
from caip_agents_sdk.observability import (
    get_current_trace_id,
    get_current_trace_url,
    score_current_trace,
)

# User ID for all traces - consistent across examples
USER_ID = "QX11111"


@observe(name="preprocess_query")
def preprocess_query(query: str) -> str:
    """Clean and normalize the user query."""
    cleaned = query.strip().lower()
    time.sleep(0.05)  # Simulate processing
    return cleaned


@observe(name="postprocess_response")
def postprocess_response(response: str) -> str:
    """Format the agent's response for display."""
    return f"📋 Agent Response:\n{'-' * 40}\n{response}"


@observe(name="simple_agent_demo", as_type="chain")
async def simple_agent_demo(query: str) -> str:
    """
    A simple traced pipeline demonstrating basic agent observability.
    
    The entire flow is wrapped in a chain observation, so all nested
    operations appear in a single consolidated trace in Langfuse.
    """

    # Propagate user context to all nested observations
    with propagate_attributes(
        user_id=USER_ID,
        session_id="session-simple-001",
        tags=["example", "simple-tracing", "production"],
        metadata={"example": "simple_agent_demo", "version": "1.0"},
    ):
        # Initialize CAIP client (auto-initializes Langfuse)
        client = CAIPAgentsClient()
        agent_id = os.getenv("CAIP_AGENT_ID", "demo-agent")
        
        # Create agent with pydantic_ai framework
        agent = client.create_agent(
            framework="pydantic_ai",
            agent_id=agent_id,
        )
        await agent.initialize()
        
        # Create conversation thread
        thread = await client.create_thread(
            agent_id=agent_id,
            thread_data={"title": "Simple Tracing Demo", "status": "open"},
        )
        agent.thread_id = thread.threadId
        print(f"✅ Agent initialized with thread: {thread.threadId}")
        
        # Register a simple tool
        @agent.tool_plain
        def get_greeting(name: str) -> str:
            """Get a personalized greeting for the user."""
            return f"Hello, {name}! Welcome to CAIP Agents SDK."
        
        # Pre-process query (traced as span)
        processed_query = preprocess_query(query)
        print(f"🔍 Processed query: {processed_query}")
        
        # Run agent (SDK auto-traces: agent → tool → LLM generation)
        result = await agent.run(processed_query)
        
        # Post-process response (traced as span)
        formatted = postprocess_response(str(result.output))
        
        # Score the trace for quality tracking
        score_current_trace(
            "user_satisfaction", 
            0.95, 
            comment="Demo completed successfully"
        )
        
        # Print trace info for debugging
        trace_id = get_current_trace_id()
        trace_url = get_current_trace_url()
        print(f"\n🔗 Trace ID: {trace_id or 'N/A'}")
        print(f"🌐 Trace URL: {trace_url or 'N/A'}")
        
        return formatted


async def main():
    """Run the simple agent tracing example."""
    enable_logging()
    
    print("=" * 60)
    print("Example 1: Simple Agent Tracing with CAIP Agents SDK")
    print("=" * 60)

    result = await simple_agent_demo(
        "  Use the get_greeting tool to greet John  "
    )
    print(f"\n{result}")

    # Flush traces before exit
    flush_traces()
    print("\n✅ All traces flushed to Langfuse")


if __name__ == "__main__":
    asyncio.run(main())

Expected Trace Tree:

trace (user: QX11111, session: session-simple-001)
  └─ chain: simple_agent_demo (200ms)
       ├─ span: preprocess_query (50ms)
       ├─ agent: caip.pydantic_ai.run (100ms)
       │    ├─ tool: caip.tool.get_greeting (10ms)
       │    └─ generation: caip.llm.call (80ms)
       └─ span: postprocess_response (10ms)

Example 2: RAG Pipeline with KnowledgeBaseSearchTool

What it demonstrates: Complete RAG (Retrieval-Augmented Generation) pipeline using CAIP's KnowledgeBaseSearchTool for semantic search with full observability.

Key Concepts:

KnowledgeBaseSearchTool for vector store search
as_type="retriever" for retrieval visualization
Hybrid search strategy (vector + keyword)
score_current_span for step-level metrics

example_rag_pipeline.py
"""
Example 2: RAG Pipeline with KnowledgeBaseSearchTool
====================================================
Retrieval-Augmented Generation using CAIP vector stores with full tracing.
"""

import asyncio
import os

from caip_agents_sdk import (
    CAIPAgentsClient,
    observe,
    propagate_attributes,
    flush_traces,
    enable_logging,
)
from caip_agents_sdk.tools import KnowledgeBaseSearchTool
from caip_agents_sdk.observability import (
    get_current_trace_id,
    get_current_trace_url,
    score_current_trace,
    score_current_span,
)

USER_ID = "QX11111"


@observe(name="knowledge_base_search", as_type="retriever")
async def search_knowledge_base(
    kb_tool: KnowledgeBaseSearchTool,
    query: str,
    top_k: int = 5
) -> list[dict]:
    """
    Search the knowledge base using hybrid search strategy.
    
    Traced as a 'retriever' for proper visualization in Langfuse.
    """
    results = await kb_tool.search(
        query=query,
        search_type="hybrid",  # Combined vector + keyword search
        top_k=top_k,
        score_threshold=0.5,
    )
    
    # Score this retrieval step
    score_current_span(
        "retrieval_quality",
        0.85 if len(results) > 0 else 0.0,
        comment=f"Retrieved {len(results)} documents"
    )
    
    return results


@observe(name="build_context")
def build_context(search_results: str) -> str:
    """Build context string from search results for the LLM."""
    context = f"""Based on the following knowledge base documents:

{search_results}

Please provide a comprehensive answer based on this context."""
    return context


@observe(name="validate_response")
def validate_response(response: str) -> bool:
    """Validate the quality of the generated response."""
    is_valid = len(response) > 20 and "error" not in response.lower()
    
    score_current_span(
        "response_validity",
        1.0 if is_valid else 0.0,
        comment="Response validation check"
    )
    
    return is_valid


@observe(name="rag_pipeline", as_type="chain")
async def rag_pipeline(query: str, vector_store_id: str) -> str:
    """
    Complete RAG pipeline with knowledge base search.
    
    The entire flow is wrapped in a chain observation for consolidated tracing.
    """

    with propagate_attributes(
        user_id=USER_ID,
        session_id="session-rag-001",
        tags=["example", "rag-pipeline", "knowledge-base"],
        metadata={"example": "rag_pipeline", "vector_store_id": vector_store_id},
    ):
        # Initialize client
        client = CAIPAgentsClient()
        agent_id = os.getenv("CAIP_AGENT_ID", "demo-agent")
        
        # Create agent with pydantic_ai
        agent = client.create_agent(
            framework="pydantic_ai",
            agent_id=agent_id,
        )
        await agent.initialize()
        
        # Create conversation thread
        thread = await client.create_thread(
            agent_id=agent_id,
            thread_data={"title": "RAG Pipeline Demo", "status": "open"},
        )
        agent.thread_id = thread.threadId
        print(f"✅ Agent initialized with thread: {thread.threadId}")
        
        # Create KnowledgeBaseSearchTool
        kb_tool = KnowledgeBaseSearchTool(
            client=client,
            vector_store_id=vector_store_id,
        )
        
        # Search knowledge base (traced as retriever)
        print(f"🔍 Searching knowledge base for: {query}")
        search_results = await search_knowledge_base(kb_tool, query, top_k=5)
        
        # Build context from results
        context = build_context(str(search_results))
        
        # Generate response using agent with context
        full_prompt = f"{context}\n\nUser question: {query}"
        result = await agent.run(full_prompt)
        
        # Validate response
        is_valid = validate_response(str(result.output))
        print(f"✓ Response valid: {is_valid}")
        
        # Score the overall trace
        score_current_trace(
            "rag_quality",
            0.9 if is_valid else 0.5,
            comment="RAG pipeline quality score"
        )
        
        # Print trace info
        trace_id = get_current_trace_id()
        trace_url = get_current_trace_url()
        print(f"\n🔗 Trace ID: {trace_id or 'N/A'}")
        print(f"🌐 Trace URL: {trace_url or 'N/A'}")
        
        return str(result.output)


async def main():
    """Run the RAG pipeline example."""
    enable_logging()
    
    print("=" * 60)
    print("Example 2: RAG Pipeline with KnowledgeBaseSearchTool")
    print("=" * 60)

    # Use your actual vector store ID
    vector_store_id = os.getenv("CAIP_VECTOR_STORE_ID", "your-vector-store-id")

    result = await rag_pipeline(
        query="What is CAIP and how does it support AI development?",
        vector_store_id=vector_store_id,
    )
    
    print(f"\n📋 Response:\n{'-' * 40}\n{result}")

    flush_traces()
    print("\n✅ All traces flushed to Langfuse")


if __name__ == "__main__":
    asyncio.run(main())

Expected Trace Tree:

trace (user: QX11111, session: session-rag-001)
  └─ chain: rag_pipeline (300ms)
       ├─ retriever: knowledge_base_search (100ms)
       ├─ span: build_context (10ms)
       ├─ agent: caip.pydantic_ai.run (150ms)
       │    └─ generation: caip.llm.call (140ms)
       └─ span: validate_response (5ms)

Example 3: Multi-Tool Agent with HTTPRequestTool

What it demonstrates: Agent with multiple tools including HTTPRequestTool for external API calls and KnowledgeBaseSearchTool for document retrieval.

Key Concepts:

HTTPRequestTool for external API calls
Multiple tool registration
Tool-level scoring with score_current_span
Allowed hosts for security

example_agent_tools.py
"""
Example 3: Multi-Tool Agent with HTTPRequestTool
=================================================
Agent with multiple tools including HTTP requests and knowledge base.
"""

import asyncio
import os

from caip_agents_sdk import (
    CAIPAgentsClient,
    observe,
    propagate_attributes,
    flush_traces,
    enable_logging,
)
from caip_agents_sdk.tools import KnowledgeBaseSearchTool, HTTPRequestTool
from caip_agents_sdk.observability import (
    get_current_trace_id,
    get_current_trace_url,
    score_current_trace,
    score_current_span,
)

USER_ID = "QX11111"


@observe(name="summarize_results")
def summarize_results(tool_outputs: list[str]) -> str:
    """Summarize outputs from multiple tools."""
    summary = f"Collected {len(tool_outputs)} tool results:\n"
    for i, output in enumerate(tool_outputs, 1):
        summary += f"  {i}. {output[:100]}...\n" if len(output) > 100 else f"  {i}. {output}\n"
    return summary


@observe(name="multi_tool_agent_demo", as_type="chain")
async def multi_tool_agent_demo(query: str, vector_store_id: str) -> str:
    """
    Multi-tool agent demonstrating combined HTTP and KB search capabilities.
    """

    with propagate_attributes(
        user_id=USER_ID,
        session_id="session-multi-tool-001",
        tags=["example", "multi-tool", "http-request", "knowledge-base"],
        metadata={"example": "multi_tool_agent_demo"},
    ):
        # Initialize client
        client = CAIPAgentsClient()
        agent_id = os.getenv("CAIP_AGENT_ID", "demo-agent")
        
        # Create agent
        agent = client.create_agent(
            framework="pydantic_ai",
            agent_id=agent_id,
        )
        await agent.initialize()
        
        # Create thread
        thread = await client.create_thread(
            agent_id=agent_id,
            thread_data={"title": "Multi-Tool Demo", "status": "open"},
        )
        agent.thread_id = thread.threadId
        print(f"✅ Agent initialized with thread: {thread.threadId}")
        
        # Create tools
        kb_tool = KnowledgeBaseSearchTool(
            client=client,
            vector_store_id=vector_store_id,
        )
        
        http_tool = HTTPRequestTool(
            allowed_hosts=["api.weatherapi.com", "api.openweathermap.org"],
            timeout=10.0,
            follow_redirects=False,
        )
        
        # Track tool outputs
        tool_outputs = []
        
        @agent.tool_plain
        async def fetch_weather_api(city: str) -> str:
            """Fetch weather data from external API."""
            try:
                result = await http_tool.execute(
                    url=f"https://api.weatherapi.com/v1/current.json?key=demo&q={city}",
                    method="GET"
                )
                tool_outputs.append(result)
                score_current_span("api_success", 1.0, comment="Weather API call successful")
                return result
            except Exception as e:
                score_current_span("api_success", 0.0, comment=f"API call failed: {e}")
                return f"Weather API error: {str(e)}"
        
        @agent.tool_plain
        async def search_documentation(search_query: str) -> str:
            """Search internal documentation using knowledge base."""
            result = await kb_tool.search(
                query=search_query,
                search_type="hybrid",
                top_k=3,
            )
            tool_outputs.append(str(result))
            score_current_span("search_success", 1.0, comment="KB search completed")
            return str(result)
        
        @agent.tool_plain
        def get_current_time() -> str:
            """Get the current timestamp."""
            from datetime import datetime
            return datetime.now().isoformat()
        
        # Run agent with multi-tool prompt
        print(f"🔍 Running agent with query: {query}")
        result = await agent.run(query)
        
        # Summarize tool usage
        summary = summarize_results(tool_outputs)
        print(f"\n📊 Tool Summary:\n{summary}")
        
        # Score the overall trace
        score_current_trace(
            "multi_tool_success",
            0.9 if len(tool_outputs) > 0 else 0.5,
            comment=f"Agent used {len(tool_outputs)} tools"
        )
        
        # Print trace info
        trace_id = get_current_trace_id()
        trace_url = get_current_trace_url()
        print(f"\n🔗 Trace ID: {trace_id or 'N/A'}")
        print(f"🌐 Trace URL: {trace_url or 'N/A'}")
        
        return str(result.output)


async def main():
    """Run the multi-tool agent example."""
    enable_logging()
    
    print("=" * 60)
    print("Example 3: Multi-Tool Agent with HTTPRequestTool")
    print("=" * 60)

    vector_store_id = os.getenv("CAIP_VECTOR_STORE_ID", "your-vector-store-id")

    result = await multi_tool_agent_demo(
        query="What time is it now? Also search the documentation for information about CAIP agents.",
        vector_store_id=vector_store_id,
    )
    
    print(f"\n📋 Response:\n{'-' * 40}\n{result}")

    flush_traces()
    print("\n✅ All traces flushed to Langfuse")


if __name__ == "__main__":
    asyncio.run(main())

Expected Trace Tree:

trace (user: QX11111, session: session-multi-tool-001)
  └─ chain: multi_tool_agent_demo (400ms)
       ├─ agent: caip.pydantic_ai.run (350ms)
       │    ├─ tool: caip.tool.get_current_time (5ms)
       │    ├─ tool: caip.tool.search_documentation (100ms)
       │    └─ generation: caip.llm.call (200ms)
       └─ span: summarize_results (10ms)

Example 4: Error Handling and Retry

What it demonstrates: Production error handling patterns with retry logic, fallback responses, and error scoring for failure tracking.

Key Concepts:

Try/except with observability context preservation
Error scoring for failure tracking
Retry logic with span tracking
Graceful degradation with fallback responses

example_error_handling.py
"""
Example 4: Error Handling and Retry with Observability
=======================================================
Production error handling patterns with full trace visibility.
"""

import asyncio
import os
import random

from caip_agents_sdk import (
    CAIPAgentsClient,
    observe,
    propagate_attributes,
    flush_traces,
    enable_logging,
)
from caip_agents_sdk.observability import (
    get_current_trace_id,
    get_current_trace_url,
    score_current_trace,
    score_current_span,
)

USER_ID = "QX11111"


class OperationError(Exception):
    """Custom exception for simulated operation failures."""
    pass


@observe(name="validate_input")
def validate_input(query: str) -> bool:
    """Validate user input before processing."""
    is_valid = len(query.strip()) > 0 and len(query) < 500
    
    score_current_span(
        "input_validity",
        1.0 if is_valid else 0.0,
        comment="Input validation check"
    )
    
    if not is_valid:
        raise ValueError("Invalid input: query must be 1-500 characters")
    
    return True


@observe(name="attempt_operation")
async def attempt_operation(operation_name: str, fail_rate: float = 0.5) -> str:
    """Simulate an operation that may fail."""
    await asyncio.sleep(0.1)
    
    if random.random() < fail_rate:
        score_current_span("operation_success", 0.0, comment=f"{operation_name} failed")
        raise OperationError(f"Operation '{operation_name}' failed (simulated)")
    
    score_current_span("operation_success", 1.0, comment=f"{operation_name} succeeded")
    return f"Operation '{operation_name}' completed successfully"


@observe(name="handle_error")
def handle_error(error: Exception, context: str) -> dict:
    """Handle and log errors with observability."""
    error_info = {
        "error_type": type(error).__name__,
        "error_message": str(error),
        "context": context,
    }
    
    score_current_span(
        "error_handled",
        1.0,
        comment=f"Handled {error_info['error_type']}: {error_info['error_message'][:50]}"
    )
    
    return error_info


@observe(name="generate_fallback")
def generate_fallback(original_query: str, error_info: dict) -> str:
    """Generate a fallback response when operation fails."""
    fallback = (
        f"I apologize, but I encountered an issue processing your request. "
        f"Error: {error_info['error_message']}. "
        f"Please try again or rephrase your question."
    )
    
    score_current_span("fallback_generated", 1.0, comment="Fallback response created")
    
    return fallback


@observe(name="retry_operation", as_type="chain")
async def retry_operation(
    operation_name: str,
    max_retries: int = 3,
    fail_rate: float = 0.7
) -> tuple[bool, str]:
    """Retry an operation with observability for each attempt."""
    last_error = None
    
    for attempt in range(1, max_retries + 1):
        try:
            print(f"  Attempt {attempt}/{max_retries}...")
            result = await attempt_operation(
                f"{operation_name}_attempt_{attempt}",
                fail_rate=fail_rate
            )
            return (True, result)
        except OperationError as e:
            last_error = e
            print(f"  Attempt {attempt} failed: {e}")
            if attempt < max_retries:
                await asyncio.sleep(0.2 * attempt)  # Exponential backoff
    
    return (False, str(last_error))


@observe(name="error_handling_demo", as_type="chain")
async def error_handling_demo(query: str) -> str:
    """Complete error handling demonstration with retry and fallback."""

    with propagate_attributes(
        user_id=USER_ID,
        session_id="session-error-001",
        tags=["example", "error-handling", "retry", "production"],
        metadata={"example": "error_handling_demo", "max_retries": "3"},
    ):
        # Validate input
        try:
            validate_input(query)
            print("✅ Input validation passed")
        except ValueError as e:
            error_info = handle_error(e, "input_validation")
            return generate_fallback(query, error_info)
        
        # Retry operation with observability
        success, result = await retry_operation(
            operation_name="data_fetch",
            max_retries=3,
            fail_rate=0.5
        )
        
        if not success:
            error_info = handle_error(
                OperationError(result),
                "retry_exhausted"
            )
            fallback = generate_fallback(query, error_info)
            score_current_trace("request_success", 0.0, comment="Request failed after retries")
            return fallback
        
        print(f"✅ Operation succeeded: {result}")
        
        # Initialize agent for processing
        try:
            client = CAIPAgentsClient()
            agent_id = os.getenv("CAIP_AGENT_ID", "demo-agent")
            
            agent = client.create_agent(
                framework="pydantic_ai",
                agent_id=agent_id,
            )
            await agent.initialize()
            
            thread = await client.create_thread(
                agent_id=agent_id,
                thread_data={"title": "Error Handling Demo", "status": "open"},
            )
            agent.thread_id = thread.threadId
            
            # Run agent
            agent_result = await agent.run(query)
            final_response = str(agent_result.output)
            
            score_current_trace(
                "request_success",
                1.0,
                comment="Request completed successfully with retries"
            )
            
        except Exception as e:
            error_info = handle_error(e, "agent_execution")
            final_response = generate_fallback(query, error_info)
            score_current_trace("request_success", 0.0, comment=f"Agent failed: {e}")
        
        # Print trace info
        trace_id = get_current_trace_id()
        trace_url = get_current_trace_url()
        print(f"\n🔗 Trace ID: {trace_id or 'N/A'}")
        print(f"🌐 Trace URL: {trace_url or 'N/A'}")
        
        return final_response


async def main():
    """Run the error handling example."""
    enable_logging()
    
    print("=" * 60)
    print("Example 4: Error Handling and Retry with Observability")
    print("=" * 60)

    # Run multiple times to see different outcomes
    for i in range(2):
        print(f"\n--- Run {i + 1} ---")
        result = await error_handling_demo(
            "What are the best practices for error handling in production?"
        )
        print(f"\n📋 Response:\n{'-' * 40}\n{result[:200]}...")

    flush_traces()
    print("\n✅ All traces flushed to Langfuse")


if __name__ == "__main__":
    asyncio.run(main())

Expected Trace Tree (with retries):

trace (user: QX11111, session: session-error-001)
  └─ chain: error_handling_demo (500ms)
       ├─ span: validate_input (10ms) ✓
       ├─ chain: retry_operation (300ms)
       │    ├─ span: attempt_operation (100ms) ✗
       │    ├─ span: attempt_operation (100ms) ✗
       │    └─ span: attempt_operation (100ms) ✓
       └─ agent: caip.pydantic_ai.run (150ms)
            └─ generation: caip.llm.call (140ms)

Example 5: Multi-Turn Conversation with Session Tracking

What it demonstrates: Session-based conversations with user attribution, multiple turns grouped by session ID, and conversation state management.

Key Concepts:

Session ID for conversation grouping
User ID for user-level analytics
Thread-based conversation state
Multiple turns in a single session

example_conversation_session.py
"""
Example 5: Multi-Turn Conversation with Session Tracking
=========================================================
Session-based conversations with user and session attribution.
"""

import asyncio
import os
from datetime import datetime

from caip_agents_sdk import (
    CAIPAgentsClient,
    observe,
    propagate_attributes,
    flush_traces,
    enable_logging,
)
from caip_agents_sdk.observability import (
    get_current_trace_id,
    get_current_trace_url,
    score_current_trace,
)

USER_ID = "QX11111"


class ConversationState:
    """Manages conversation history and context."""
    
    def __init__(self, session_id: str, user_id: str):
        self.session_id = session_id
        self.user_id = user_id
        self.history: list[dict] = []
        self.turn_count = 0
        self.start_time = datetime.now()
    
    def add_turn(self, user_message: str, assistant_response: str):
        """Add a conversation turn to history."""
        self.turn_count += 1
        self.history.append({
            "turn": self.turn_count,
            "timestamp": datetime.now().isoformat(),
            "user": user_message,
            "assistant": assistant_response,
        })
    
    def get_context(self) -> str:
        """Get conversation context for the next turn."""
        if not self.history:
            return "This is the start of a new conversation."
        
        context = "Previous conversation:\n"
        for turn in self.history[-3:]:  # Last 3 turns for context
            context += f"User: {turn['user']}\n"
            context += f"Assistant: {turn['assistant']}\n\n"
        return context


@observe(name="prepare_context")
def prepare_context(state: ConversationState, new_message: str) -> str:
    """Prepare context for the next conversation turn."""
    history_context = state.get_context()
    full_context = f"""Session: {state.session_id}
Turn: {state.turn_count + 1}
User: {state.user_id}

{history_context}

Current user message: {new_message}

Please respond helpfully and maintain conversation continuity."""
    
    return full_context


@observe(name="update_history")
def update_history(
    state: ConversationState,
    user_message: str,
    assistant_response: str
) -> None:
    """Update conversation history with the latest turn."""
    state.add_turn(user_message, assistant_response)


@observe(name="conversation_turn", as_type="chain")
async def conversation_turn(
    client: CAIPAgentsClient,
    agent,
    state: ConversationState,
    user_message: str,
) -> str:
    """Process a single conversation turn with full tracing."""

    with propagate_attributes(
        user_id=state.user_id,
        session_id=state.session_id,
        tags=["example", "conversation", f"turn-{state.turn_count + 1}"],
        metadata={
            "turn_number": str(state.turn_count + 1),
            "session_start": state.start_time.isoformat(),
        },
    ):
        # Prepare context
        context = prepare_context(state, user_message)
        
        # Run agent
        print(f"  🔍 Processing: '{user_message[:50]}...'")
        result = await agent.run(context)
        response = str(result.output)
        
        # Update history
        update_history(state, user_message, response)
        
        # Score this turn
        score_current_trace(
            "turn_quality",
            0.85,
            comment=f"Turn {state.turn_count} completed"
        )
        
        return response


@observe(name="multi_turn_session", as_type="chain")
async def multi_turn_session(user_messages: list[str]) -> dict:
    """Run a complete multi-turn conversation session."""
    session_id = f"session-conv-{datetime.now().strftime('%Y%m%d-%H%M%S')}"
    
    with propagate_attributes(
        user_id=USER_ID,
        session_id=session_id,
        tags=["example", "multi-turn", "session"],
        metadata={"total_turns": str(len(user_messages))},
    ):
        # Initialize client and agent
        client = CAIPAgentsClient()
        agent_id = os.getenv("CAIP_AGENT_ID", "demo-agent")
        
        agent = client.create_agent(
            framework="pydantic_ai",
            agent_id=agent_id,
        )
        await agent.initialize()
        
        # Create thread for the session
        thread = await client.create_thread(
            agent_id=agent_id,
            thread_data={
                "title": f"Multi-Turn Session {session_id}",
                "status": "open",
            },
        )
        agent.thread_id = thread.threadId
        print(f"✅ Session started: {session_id}")
        print(f"   Thread ID: {thread.threadId}")
        
        # Initialize conversation state
        state = ConversationState(session_id=session_id, user_id=USER_ID)
        
        # Process each turn
        responses = []
        for i, message in enumerate(user_messages, 1):
            print(f"\n📝 Turn {i}/{len(user_messages)}")
            response = await conversation_turn(client, agent, state, message)
            responses.append(response)
            print(f"   Response: {response[:100]}...")
        
        # Score the overall session
        score_current_trace(
            "session_quality",
            0.9,
            comment=f"Session completed with {len(user_messages)} turns"
        )
        
        # Print trace info
        trace_id = get_current_trace_id()
        trace_url = get_current_trace_url()
        print(f"\n🔗 Session Trace ID: {trace_id or 'N/A'}")
        print(f"🌐 Session Trace URL: {trace_url or 'N/A'}")
        
        return {
            "session_id": session_id,
            "user_id": USER_ID,
            "total_turns": state.turn_count,
            "history": state.history,
            "responses": responses,
        }


async def main():
    """Run the multi-turn conversation example."""
    enable_logging()
    
    print("=" * 60)
    print("Example 5: Multi-Turn Conversation with Session Tracking")
    print("=" * 60)

    # Simulate a multi-turn conversation
    user_messages = [
        "Hello! I'm interested in learning about CAIP Agents SDK.",
        "What are the main features it provides?",
        "How do I get started with building my first agent?",
        "Thanks for the information! That's very helpful.",
    ]

    result = await multi_turn_session(user_messages)
    
    print(f"\n📊 Session Summary:")
    print(f"   Session ID: {result['session_id']}")
    print(f"   User ID: {result['user_id']}")
    print(f"   Total Turns: {result['total_turns']}")

    flush_traces()
    print("\n✅ All traces flushed to Langfuse")


if __name__ == "__main__":
    asyncio.run(main())

Expected Trace Tree:

trace (user: QX11111, session: session-conv-20240115-143022)
  └─ chain: multi_turn_session (800ms)
       ├─ chain: conversation_turn (turn 1) (200ms)
       │    ├─ span: prepare_context (10ms)
       │    ├─ agent: caip.pydantic_ai.run (180ms)
       │    └─ span: update_history (5ms)
       ├─ chain: conversation_turn (turn 2) (200ms)
       │    └─ ...
       ├─ chain: conversation_turn (turn 3) (200ms)
       │    └─ ...
       └─ chain: conversation_turn (turn 4) (200ms)
            └─ ...

Example 6: Custom Metrics and Quality Scoring

What it demonstrates: Production monitoring with custom scores, evaluation metrics, quality gates, and comprehensive trace scoring.

Key Concepts:

score_current_trace for trace-level metrics
score_current_span for step-level metrics
as_type="evaluator" for evaluation visibility
Quality gates with scoring thresholds
Performance metrics collection

example_custom_metrics.py
"""
Example 6: Custom Metrics and Quality Scoring
==============================================
Production monitoring with custom scores and evaluation metrics.
"""

import asyncio
import os
import time
from dataclasses import dataclass

from caip_agents_sdk import (
    CAIPAgentsClient,
    observe,
    propagate_attributes,
    flush_traces,
    enable_logging,
)
from caip_agents_sdk.observability import (
    get_current_trace_id,
    get_current_trace_url,
    score_current_trace,
    score_current_span,
)

USER_ID = "QX11111"


@dataclass
class MetricsResult:
    """Container for collected metrics."""
    latency_ms: float
    quality_score: float
    safety_score: float
    response_length: int
    word_count: int
    overall_score: float


@observe(name="measure_latency")
async def measure_latency(func_name: str, duration_ms: float) -> None:
    """Record latency metric for a function."""
    latency_score = max(0, 1.0 - (duration_ms / 5000))  # 5s = 0 score
    
    score_current_span(
        "latency_ms",
        duration_ms,
        comment=f"{func_name} took {duration_ms:.2f}ms"
    )
    
    score_current_span(
        "latency_score",
        latency_score,
        comment=f"Latency score (lower latency = higher score)"
    )


@observe(name="evaluate_response_quality", as_type="evaluator")
def evaluate_response_quality(response: str) -> float:
    """Evaluate response quality using heuristics."""
    quality_score = 0.0
    reasons = []
    
    # Length check
    length = len(response)
    if 50 <= length <= 2000:
        quality_score += 0.3
        reasons.append("good_length")
    
    # Completeness check
    if response.strip().endswith(('.', '!', '?')):
        quality_score += 0.2
        reasons.append("complete_sentence")
    
    # No error indicators
    if "error" not in response.lower() and "sorry" not in response.lower():
        quality_score += 0.3
        reasons.append("no_error_indicators")
    
    # Has substantive content
    word_count = len(response.split())
    if word_count >= 10:
        quality_score += 0.2
        reasons.append("substantive_content")
    
    score_current_span(
        "quality_score",
        quality_score,
        comment=f"Quality factors: {', '.join(reasons)}"
    )
    
    return quality_score


@observe(name="evaluate_safety", as_type="evaluator")
def evaluate_safety(response: str) -> float:
    """Evaluate response safety using basic checks."""
    safety_score = 1.0
    issues = []
    
    sensitive_patterns = [
        "password", "secret", "api_key", "token",
        "credit card", "ssn", "social security"
    ]
    
    response_lower = response.lower()
    for pattern in sensitive_patterns:
        if pattern in response_lower:
            safety_score -= 0.3
            issues.append(f"contains_{pattern.replace(' ', '_')}")
    
    safety_score = max(0.0, safety_score)
    
    score_current_span(
        "safety_score",
        safety_score,
        comment=f"Issues: {', '.join(issues) if issues else 'none'}"
    )
    
    return safety_score


@observe(name="aggregate_metrics")
def aggregate_metrics(
    latency_ms: float,
    quality_score: float,
    safety_score: float,
    response: str,
) -> MetricsResult:
    """Aggregate all metrics into a final result."""
    overall_score = (
        quality_score * 0.4 +
        safety_score * 0.4 +
        max(0, 1.0 - latency_ms / 5000) * 0.2
    )
    
    result = MetricsResult(
        latency_ms=latency_ms,
        quality_score=quality_score,
        safety_score=safety_score,
        response_length=len(response),
        word_count=len(response.split()),
        overall_score=overall_score,
    )
    
    score_current_span(
        "overall_score",
        overall_score,
        comment=f"Aggregated from quality={quality_score:.2f}, safety={safety_score:.2f}"
    )
    
    return result


@observe(name="metrics_pipeline", as_type="chain")
async def metrics_pipeline(query: str) -> dict:
    """Complete metrics collection pipeline with quality scoring."""

    with propagate_attributes(
        user_id=USER_ID,
        session_id="session-metrics-001",
        tags=["example", "metrics", "quality-scoring", "production"],
        metadata={"example": "metrics_pipeline"},
    ):
        # Initialize client
        client = CAIPAgentsClient()
        agent_id = os.getenv("CAIP_AGENT_ID", "demo-agent")
        
        # Create agent
        agent = client.create_agent(
            framework="pydantic_ai",
            agent_id=agent_id,
        )
        await agent.initialize()
        
        # Create thread
        thread = await client.create_thread(
            agent_id=agent_id,
            thread_data={"title": "Metrics Demo", "status": "open"},
        )
        agent.thread_id = thread.threadId
        print(f"✅ Agent initialized with thread: {thread.threadId}")
        
        # Run agent and measure latency
        print(f"🔍 Processing query: '{query[:50]}...'")
        start_time = time.time()
        result = await agent.run(query)
        end_time = time.time()
        
        latency_ms = (end_time - start_time) * 1000
        response = str(result.output)
        
        # Record latency
        await measure_latency("agent_run", latency_ms)
        print(f"⏱️  Latency: {latency_ms:.2f}ms")
        
        # Evaluate quality (traced as evaluator)
        quality_score = evaluate_response_quality(response)
        print(f"📊 Quality Score: {quality_score:.2f}")
        
        # Evaluate safety (traced as evaluator)
        safety_score = evaluate_safety(response)
        print(f"🛡️  Safety Score: {safety_score:.2f}")
        
        # Aggregate metrics
        metrics = aggregate_metrics(
            latency_ms=latency_ms,
            quality_score=quality_score,
            safety_score=safety_score,
            response=response,
        )
        print(f"📈 Overall Score: {metrics.overall_score:.2f}")
        
        # Score the overall trace
        score_current_trace("quality", quality_score, comment="Response quality")
        score_current_trace("safety", safety_score, comment="Response safety")
        score_current_trace("latency_ms", latency_ms, comment="Response latency")
        score_current_trace("overall", metrics.overall_score, comment="Overall score")
        
        # Quality gate check
        if metrics.overall_score < 0.5:
            score_current_trace(
                "quality_gate",
                0.0,
                comment="FAILED - Overall score below threshold"
            )
            print("⚠️  Quality Gate: FAILED")
        else:
            score_current_trace(
                "quality_gate",
                1.0,
                comment="PASSED - Quality gate passed"
            )
            print("✅ Quality Gate: PASSED")
        
        # Print trace info
        trace_id = get_current_trace_id()
        trace_url = get_current_trace_url()
        print(f"\n🔗 Trace ID: {trace_id or 'N/A'}")
        print(f"🌐 Trace URL: {trace_url or 'N/A'}")
        
        return {
            "response": response,
            "metrics": {
                "latency_ms": metrics.latency_ms,
                "quality_score": metrics.quality_score,
                "safety_score": metrics.safety_score,
                "response_length": metrics.response_length,
                "word_count": metrics.word_count,
                "overall_score": metrics.overall_score,
            },
            "trace_id": trace_id,
        }


async def main():
    """Run the metrics collection example."""
    enable_logging()
    
    print("=" * 60)
    print("Example 6: Custom Metrics and Quality Scoring")
    print("=" * 60)

    result = await metrics_pipeline(
        "Explain the benefits of using observability in production AI systems. "
        "Include specific examples of metrics that should be tracked."
    )
    
    print(f"\n📋 Response:\n{'-' * 40}\n{result['response'][:300]}...")
    print(f"\n📊 Metrics Summary:")
    for key, value in result['metrics'].items():
        print(f"   {key}: {value:.2f}" if isinstance(value, float) else f"   {key}: {value}")

    flush_traces()
    print("\n✅ All traces flushed to Langfuse")


if __name__ == "__main__":
    asyncio.run(main())

Expected Trace Tree:

trace (user: QX11111, session: session-metrics-001)
  └─ chain: metrics_pipeline (400ms)
       ├─ span: measure_latency (5ms)
       ├─ agent: caip.pydantic_ai.run (300ms)
       │    └─ generation: caip.llm.call (280ms)
       ├─ evaluator: evaluate_response_quality (10ms)
       ├─ evaluator: evaluate_safety (10ms)
       └─ span: aggregate_metrics (5ms)

Scores attached:
├── quality: 0.85
├── safety: 1.0
├── latency_ms: 300
├── overall: 0.89
└── quality_gate: 1.0 (PASSED)

Running the Examples

# Set up environment
cd your-project
python -m venv .venv
source .venv/bin/activate

# Install dependencies
pip install caip-agents-sdk python-dotenv \
  --index-url https://packages.orbit.bmwgroup.net/artifactory/api/pypi/connected-ai-platform-pypi-local-public/simple \
  --extra-index-url https://pypi.org/simple

# Create .env file with your Langfuse credentials
cat > .env << EOF
LANGFUSE_PUBLIC_KEY=pk-lf-your-key
LANGFUSE_SECRET_KEY=sk-lf-your-key
LANGFUSE_HOST=https://langfuse.caip.bmw.cloud
LANGFUSE_ENV=prod
CAIP_AGENT_ID=your-agent-id
CAIP_REGION=ROW
# Optional: disable Langfuse observability
# CAIP_LANGFUSE_OBSERVABILITY=false
CAIP_VECTOR_STORE_ID=your-vector-store-id  # For RAG examples
EOF

# Run any example
python example_simple_tracing.py        # Example 1: Simple Agent Tracing
python example_rag_pipeline.py          # Example 2: RAG Pipeline
python example_agent_tools.py           # Example 3: Multi-Tool Agent
python example_error_handling.py        # Example 4: Error Handling & Retry
python example_conversation_session.py  # Example 5: Multi-Turn Conversation
python example_custom_metrics.py        # Example 6: Custom Metrics & Scoring

Summary of Key Patterns

Pattern	Observation Type	Use Case
`@observe(as_type="chain")`	Chain	Consolidate entire pipeline in one trace
`@observe(as_type="retriever")`	Retriever	Vector/document search operations
`@observe(as_type="evaluator")`	Evaluator	Quality/safety evaluation steps
`@observe(as_type="tool")`	Tool	Tool/function invocations
`propagate_attributes(user_id=...)`	Context	User/session attribution
`score_current_trace(...)`	Score	Trace-level metrics
`score_current_span(...)`	Score	Step-level metrics

Next Steps

Tracing Guide — Deep dive into trace instrumentation
UI Guide — Navigate the Langfuse dashboard
Best Practices — Production-ready patterns

Example 1: Simple Agent Tracing​

Example 2: RAG Pipeline with KnowledgeBaseSearchTool​

Example 3: Multi-Tool Agent with HTTPRequestTool​

Example 4: Error Handling and Retry​

Example 5: Multi-Turn Conversation with Session Tracking​

Example 6: Custom Metrics and Quality Scoring​

Running the Examples​

Summary of Key Patterns​

Next Steps​

Example 1: Simple Agent Tracing

Example 2: RAG Pipeline with KnowledgeBaseSearchTool

Example 3: Multi-Tool Agent with HTTPRequestTool

Example 4: Error Handling and Retry

Example 5: Multi-Turn Conversation with Session Tracking

Example 6: Custom Metrics and Quality Scoring

Running the Examples

Summary of Key Patterns

Next Steps