Logging & Observability | Smallest AI Docs

Good logging helps you understand what your agent is doing. In production, observability is essential for debugging issues and improving performance.

Basic Logging with Loguru

The Atoms Agent Crews SDK uses Loguru for logging. Set it up in your agent:

from loguru import logger
class MyAgent(OutputCrewNode):
    async def generate_response(self):
        logger.info("Generating response")
        
        try:
            response = await self.llm.chat(
                messages=self.context.messages,
                stream=True
            )
            
            async for chunk in response:
                if chunk.content:
                    yield chunk.content
                    
        except Exception as e:
            logger.error(f"LLM call failed: {e}")
            yield "I'm having trouble right now. Please try again."

Log Levels

Use appropriate log levels:

Level	Use For	Example
DEBUG	Detailed debugging info	Message contents, event data
INFO	Normal operations	Session start, tool calls
WARNING	Unexpected but handled	Fallback used, retry needed
ERROR	Failures	API errors, exceptions

logger.debug(f"Received event: {event}")
logger.info(f"Session started: {session.id}")
logger.warning(f"Using fallback LLM, primary failed")
logger.error(f"Tool execution failed: {error}")

Structured Logging

Add context to your logs:

logger.info(
    "Tool executed",
    extra={
        "tool_name": "get_order",
        "duration_ms": 145,
        "success": True,
        "session_id": self.session_id
    }
)

Logging Events

Track all events flowing through your agent:

class LoggingAgent(OutputCrewNode):
    async def process_event(self, event):
        logger.debug(
            f"Processing event",
            extra={
                "event_type": event.type,
                "session_id": self.session_id
            }
        )
        
        await super().process_event(event)

Logging Tool Calls

Track tool usage:

async def generate_response(self):
    # ... LLM call ...
    
    if tool_calls:
        for call in tool_calls:
            logger.info(
                f"Executing tool: {call.name}",
                extra={
                    "tool_name": call.name,
                    "arguments": call.args
                }
            )
        
        start = time.time()
        results = await self.tool_registry.execute(tool_calls, parallel=True)
        duration = time.time() - start
        
        logger.info(
            f"Tools executed in {duration:.2f}s",
            extra={
                "tool_count": len(tool_calls),
                "duration_seconds": duration
            }
        )

Session Logging

Log session lifecycle:

async def setup(session: CrewSession):
    logger.info(f"Session started: {session.id}")
    
    agent = MyAgent()
    session.add_node(agent)
    await session.start()
    
    try:
        await session.wait_until_complete()
    finally:
        logger.info(f"Session ended: {session.id}")

Log Output Configuration

Configure log output format:

import sys
from loguru import logger
# Remove default handler
logger.remove()
# Add custom handler
logger.add(
    sys.stderr,
    format="{time:HH:mm:ss} | {level} | {message}",
    level="INFO"
)
# Add file handler for debugging
logger.add(
    "logs/agent_{time}.log",
    rotation="100 MB",
    retention="7 days",
    level="DEBUG"
)

Performance Tracking

Measure response times:

import time
class TimedAgent(OutputCrewNode):
    async def generate_response(self):
        start = time.time()
        chunk_count = 0
        
        async for chunk in self._generate():
            chunk_count += 1
            yield chunk
        
        duration = time.time() - start
        logger.info(
            f"Response generated",
            extra={
                "duration_seconds": duration,
                "chunk_count": chunk_count,
                "avg_chunk_time": duration / chunk_count if chunk_count else 0
            }
        )

Error Tracking

Capture and log errors with context:

async def generate_response(self):
    try:
        async for chunk in self._generate():
            yield chunk
            
    except Exception as e:
        logger.exception(
            "Response generation failed",
            extra={
                "error_type": type(e).__name__,
                "session_id": self.session_id,
                "message_count": len(self.context.messages)
            }
        )
        yield "I encountered an error. Let me try again."

Metrics Collection

Track key metrics for dashboards:

from prometheus_client import Counter, Histogram
# Define metrics
tool_calls = Counter("agent_tool_calls_total", "Total tool calls", ["tool_name"])
response_time = Histogram("agent_response_seconds", "Response generation time")
class MetricsAgent(OutputCrewNode):
    async def generate_response(self):
        with response_time.time():
            # ... response generation ...
            pass
    
    @function_tool()
    def get_order(self, order_id: str) -> dict:
        tool_calls.labels(tool_name="get_order").inc()
        # ... implementation ...

Debug Mode

Add a debug mode for development:

import os
DEBUG = os.getenv("DEBUG", "false").lower() == "true"
class MyAgent(OutputCrewNode):
    async def generate_response(self):
        if DEBUG:
            logger.debug(f"Context: {self.context.messages}")
        
        # ... normal generation ...

Enable with:

DEBUG=true python agent.py

Next Steps

Common Issues

Troubleshoot common problems.

Examples Gallery

See complete working examples.