Laminar Tracing Integration

What is Laminar?

Laminar is an observability platform specifically designed for LLM applications. It provides:

Trace visualization - See every LLM call with full prompts and responses
Cost tracking - Monitor token usage and API costs
Performance analysis - Identify slow calls and bottlenecks
Accuracy verification - Detect hallucinations and errors
Agent debugging - Track multi-step agent workflows

Get Started with Laminar

Setup

Installation

Laminar SDK is included in JARVIS dependencies:

pyproject.toml

[project]
dependencies = [
  "lmnr>=0.7.40",
  # ... other dependencies
]

Configuration

Set your Laminar API key:

.env

LMNR_PROJECT_API_KEY=your-api-key-here

Initialization

Laminar is automatically initialized at application startup:

backend/observability/laminar.py

from lmnr import Laminar
from config import Settings
from loguru import logger

_initialized = False

def initialize_laminar(settings: Settings) -> bool:
    """Initialize Laminar SDK for tracing. Call once at app startup.
    
    Returns True if Laminar was successfully initialized, False otherwise.
    """
    global _initialized
    if _initialized:
        return True
    
    if not settings.laminar_api_key:
        logger.warning("LMNR_PROJECT_API_KEY not set — Laminar tracing disabled")
        return False
    
    try:
        Laminar.initialize(project_api_key=settings.laminar_api_key)
        _initialized = True
        logger.info("Laminar tracing initialized")
        return True
    except Exception as exc:
        logger.error("Failed to initialize Laminar: {}", exc)
        return False

Tracing Functions

Basic @observe Decorator

Use the @observe decorator from the Laminar SDK:

from lmnr import observe

@observe()
async def identify_person(image_bytes: bytes) -> dict:
    """Traced: PimEyes search + Gemini extraction."""
    # Every function call is traced
    results = await pimeyes_search(image_bytes)
    identity = await gemini_extract(results)
    return identity

This automatically:

Creates a trace span
Records start/end timestamps
Captures function arguments and return values
Sends to Laminar dashboard

Enhanced @traced Decorator

JARVIS provides an enhanced decorator that combines logging and tracing:

backend/observability/laminar.py

from observability.laminar import traced

@traced(
    name="synthesize_report",
    metadata={"llm": "gemini-2.0-flash"},
    tags=["synthesis", "critical-path"],
)
async def synthesize_report(person_id: str) -> dict:
    """Traced: Report synthesis from all intel fragments."""
    fragments = await get_fragments(person_id)
    report = await llm_synthesize(fragments)
    return report

The @traced decorator:

Logs start/end with duration to console
Sends spans to Laminar (if configured)
Works with both sync and async functions
Includes custom metadata and tags
Gracefully degrades if Laminar is not configured

Implementation

backend/observability/laminar.py

import functools
import time
from typing import Any, Callable
from loguru import logger

def traced(
    name: str,
    metadata: dict[str, Any] | None = None,
    *,
    tags: list[str] | None = None,
) -> Callable:
    """Lightweight timing + logging decorator that also feeds Laminar spans.
    
    Always logs duration/status even without Laminar.
    Enhanced metadata is merged into spans for filtering in the Laminar dashboard.
    """
    def decorator(fn):
        import asyncio
        
        if asyncio.iscoroutinefunction(fn):
            @functools.wraps(fn)
            async def async_wrapper(*args, **kwargs):
                start = time.monotonic()
                logger.debug("trace.start name={}", name)
                try:
                    result = await fn(*args, **kwargs)
                    elapsed = time.monotonic() - start
                    logger.info(
                        "trace.end name={} status=ok elapsed={:.2f}s",
                        name, elapsed
                    )
                    return result
                except Exception as exc:
                    elapsed = time.monotonic() - start
                    logger.error(
                        "trace.end name={} status=error elapsed={:.2f}s error={}",
                        name, elapsed, exc,
                    )
                    raise
            
            # Layer Laminar observe on top if available
            wrapped = async_wrapper
            if _initialized:
                try:
                    from lmnr import observe
                    span_meta = dict(metadata or {})
                    if tags:
                        span_meta["tags"] = tags
                    wrapped = observe(name=name, metadata=span_meta)(async_wrapper)
                except Exception:
                    pass
            return wrapped
        else:
            # Similar implementation for sync functions
            ...
    
    return decorator

Real-World Examples

Tracing Face Identification

backend/identification/pimeyes.py

from observability.laminar import traced

class PimEyesSearcher:
    @traced(
        name="pimeyes_search",
        metadata={"service": "pimeyes"},
        tags=["face-search", "critical-path"],
    )
    async def search(self, face_image_bytes: bytes) -> dict:
        """Run PimEyes search and extract identity info."""
        account = self._rotate_account()
        
        # Browser Use agent to navigate PimEyes
        browser = Browser(config=BrowserConfig(headless=True))
        agent = Agent(
            task=f"""Login to PimEyes and search for face...""",
            llm="gemini-2.0-flash",
            browser=browser,
        )
        
        result = await agent.run()
        screenshot = result.screenshot()
        await browser.close()
        
        # Parse screenshot with Gemini
        identity = await self._parse_results(screenshot)
        return identity
    
    @traced(
        name="pimeyes_parse_results",
        metadata={"llm": "gemini-2.0-flash"},
        tags=["vision-llm"],
    )
    async def _parse_results(self, screenshot_bytes: bytes) -> dict:
        """Use Gemini 2.0 Flash to extract identity from PimEyes results."""
        image_b64 = base64.b64encode(screenshot_bytes).decode()
        
        response = self.model.generate_content([
            {"mime_type": "image/png", "data": image_b64},
            """Analyze this PimEyes search results page..."""
        ])
        
        return json.loads(response.text)

Tracing Agent Swarm

backend/agents/orchestrator.py

from observability.laminar import traced

class SwarmOrchestrator:
    @traced(
        name="agent_swarm_execute",
        metadata={"agent_count": 4},
        tags=["agent-swarm", "research"],
    )
    async def execute(self) -> dict:
        """Run full two-tier research pipeline."""
        # Update status
        await self.convex.mutation("persons:updateStatus", {
            "personId": self.person_id,
            "status": "researching"
        })
        
        # TIER 1: Fast API enrichment
        exa_context = await self._tier1_enrichment()
        
        # TIER 2: Deep browser research
        agents = self._create_agents(exa_context)
        
        try:
            results = await asyncio.wait_for(
                asyncio.gather(
                    *[self._run_agent(agent) for agent in agents],
                    return_exceptions=True
                ),
                timeout=180  # 3 minutes
            )
        except asyncio.TimeoutError:
            results = []  # Partial results already streamed
        
        # Trigger synthesis
        return await self._synthesize()
    
    @traced(
        name="agent_execute",
        metadata={"agent_type": "linkedin"},
        tags=["browser-agent"],
    )
    async def _run_agent(self, agent) -> dict:
        """Run single agent and stream results to Convex."""
        try:
            result = await agent.run()
            await self._store_fragment(
                source=agent.SOURCE_NAME,
                data_type="profile",
                content=result
            )
            return result
        except Exception as e:
            await self._store_fragment(
                source=agent.SOURCE_NAME,
                data_type="error",
                content={"error": str(e)}
            )
            return {"error": str(e)}

Tracing Report Synthesis

backend/synthesis/synthesizer.py

from observability.laminar import traced

class ReportSynthesizer:
    @traced(
        name="synthesize_dossier",
        metadata={"llm": "gemini-2.0-flash"},
        tags=["synthesis", "critical-path"],
    )
    async def synthesize(self, person_id: str) -> dict:
        """Aggregate all fragments and generate structured dossier."""
        # Get all fragments for this person
        fragments = await self.convex.query("intelFragments:byPerson", {
            "personId": person_id
        })
        
        # Build context from fragments
        fragment_text = "\n\n".join([
            f"Source: {f['source']} ({f['dataType']})\n{f['content']}"
            for f in fragments
        ])
        
        # Synthesize with Gemini
        response = self.model.generate_content(f"""
        You are an intelligence analyst. Given the following raw data fragments
        about a person, synthesize them into a structured dossier...
        
        RAW DATA:
        {fragment_text}
        """)
        
        dossier = json.loads(response.text)
        
        # Update person record in Convex
        await self.convex.mutation("persons:updateDossier", {
            "personId": person_id,
            "dossier": dossier,
            "status": "complete"
        })
        
        return dossier

Viewing Traces

Laminar Dashboard

Access your traces at app.lmnr.ai:

Traces View - See all traces with duration and status
Trace Detail - Click a trace to see full span tree
LLM Calls - View prompts, responses, and token usage
Filters - Filter by metadata tags, status, duration

Trace Hierarchy

A typical JARVIS trace shows:

Capture Pipeline (3.2s) ✓
├── frame_extraction (0.1s) ✓
├── face_detection (0.3s) ✓
├── pimeyes_search (1.2s) ✓
│   ├── browser_navigation (0.8s) ✓
│   └── screenshot_capture (0.4s) ✓
├── pimeyes_parse_results (0.9s) ✓
│   └── gemini-2.0-flash (0.8s, 1,234 tokens, $0.002)
└── agent_swarm_execute (2.1s) ✓
    ├── agent_execute [linkedin] (1.8s) ✓
    ├── agent_execute [twitter] (1.2s) ✓
    └── tier1_enrichment [exa] (0.3s) ✓

LLM Call Details

Click on an LLM span to see:

Full prompt - Exact prompt sent to the model
Response - Complete response from the model
Tokens - Prompt tokens, completion tokens, total
Cost - Estimated API cost
Latency - Time to first token, total duration
Model - Model name and version

Filtering and Analysis

Filter by Tags

Use tags to filter traces:

@traced(
    name="linkedin_research",
    tags=["agent", "linkedin", "critical-path"],
)

In Laminar dashboard:

Filter by tag: tags:critical-path
Show only agent calls: tags:agent
Show errors: status:error

Filter by Metadata

Add metadata for richer filtering:

@traced(
    name="agent_execute",
    metadata={
        "agent_type": "linkedin",
        "person_id": person_id,
        "llm": "gemini-2.0-flash",
    },
)

Filter in dashboard:

metadata.agent_type:linkedin
metadata.llm:gemini*

Performance Analysis

Identify slow calls: Sort by duration
Token usage: Sum tokens by model
Error rate: Filter by status:error
Cost tracking: View total API costs

Best Practices

Trace all LLM calls

Every function that calls an LLM should be traced:

@traced(name="vision_extract", tags=["llm"])
async def extract_identity(image: bytes):
    response = await openai.chat.completions.create(...)
    return response

Use descriptive names

Name traces clearly to identify them in the dashboard:

# Good
@traced(name="linkedin_profile_extract")

# Bad
@traced(name="extract")

Add contextual metadata

Include IDs and parameters for debugging:

@traced(
    name="synthesize_report",
    metadata={
        "person_id": person_id,
        "fragment_count": len(fragments),
        "llm": "gemini-2.0-flash",
    },
)

Tag critical paths

Use tags to identify important traces:

@traced(
    name="capture_pipeline",
    tags=["critical-path", "user-facing"],
)

Graceful degradation

Always design tracing to degrade gracefully:

# The @traced decorator falls back to logging-only if Laminar fails
@traced(name="my_function")
async def my_function():
    # Works even if Laminar is not configured
    pass

Troubleshooting

Traces Not Appearing

Check API key

echo $LMNR_PROJECT_API_KEY

Verify initialization

Check startup logs for:

INFO: Laminar tracing initialized

Or warning:

WARNING: LMNR_PROJECT_API_KEY not set — Laminar tracing disabled

Test with simple trace

from lmnr import observe

@observe()
def test_trace():
    return "hello"

test_trace()

Check app.lmnr.ai for the trace.

Check network connectivity

Ensure your application can reach Laminar’s API:

curl https://api.lmnr.ai/health

High Overhead

If tracing adds too much latency:

Sample traces: Only trace 10% of requests

import random

if random.random() < 0.1:
    @traced(name="operation")
    async def operation():
        pass

Disable in production: Set LMNR_PROJECT_API_KEY= to disable
Reduce span depth: Only trace top-level functions

Missing Context

If traces lack context:

# Add more metadata
@traced(
    name="operation",
    metadata={
        "user_id": user_id,
        "request_id": request_id,
        "environment": settings.environment,
    },
)

Next Steps

Observability

Learn about the full observability stack

Performance

Optimize performance based on traces

Deployment

Deploy with production tracing

Architecture

Understand system architecture

Setup

Advanced

Contributing

Laminar Tracing Integration

What is Laminar?

Get Started with Laminar

Setup

Installation

Configuration

Initialization

Tracing Functions

Basic @observe Decorator

Enhanced @traced Decorator

Implementation

Real-World Examples

Tracing Face Identification

Tracing Agent Swarm

Tracing Report Synthesis

Viewing Traces

Laminar Dashboard

Trace Hierarchy

LLM Call Details

Filtering and Analysis

Filter by Tags

Filter by Metadata

Performance Analysis

Best Practices

Troubleshooting

Traces Not Appearing

High Overhead

Missing Context

Next Steps

Observability

Performance

Deployment

Architecture

Build docs developers (and LLMs) love

Setup

Advanced

Contributing

​What is Laminar?

Get Started with Laminar

​Setup

​Installation

​Configuration

​Initialization

​Tracing Functions

​Basic @observe Decorator

​Enhanced @traced Decorator

​Implementation

​Real-World Examples

​Tracing Face Identification

​Tracing Agent Swarm

​Tracing Report Synthesis

​Viewing Traces

​Laminar Dashboard

​Trace Hierarchy

​LLM Call Details

​Filtering and Analysis

​Filter by Tags

​Filter by Metadata

​Performance Analysis

​Best Practices

​Troubleshooting

​Traces Not Appearing

​High Overhead

​Missing Context

​Next Steps

Observability

Performance

Deployment

Architecture

Build docs developers (and LLMs) love

What is Laminar?

Setup

Installation

Configuration

Initialization

Tracing Functions

Basic @observe Decorator

Enhanced @traced Decorator

Implementation

Real-World Examples

Tracing Face Identification

Tracing Agent Swarm

Tracing Report Synthesis

Viewing Traces

Laminar Dashboard

Trace Hierarchy

LLM Call Details

Filtering and Analysis

Filter by Tags

Filter by Metadata

Performance Analysis

Best Practices

Troubleshooting

Traces Not Appearing

High Overhead

Missing Context

Next Steps