Tracing

Tracing is the observability layer of Agentswarm. It allows you to peer inside the "black box" of agent execution, monitoring not just final outputs but the internal reasoning, tool calls, and state changes.

The Abstract Concept

The Tracing class is an abstract contract (interface). The core framework makes calls to this interface at key moments: - When an agent starts execution. - When an agent finishes or errors. - When a loop step (like a ReAct iteration) begins. - When a tool result is received.

By implementing this interface, you can route these signals to any observability backend you prefer.

agentswarm.utils.tracing.Tracing

Bases: ABC

Source code in src/agentswarm/utils/tracing.py

class Tracing(ABC):
    @abstractmethod
    def trace_agent(self, context: Context, agent_id: str, arguments: dict):
        pass

    @abstractmethod
    def trace_loop_step(self, context: Context, step_name: str):
        pass

    @abstractmethod
    def trace_agent_result(self, context: Context, agent_id: str, result: Any):
        pass

    @abstractmethod
    def trace_agent_error(self, context: Context, agent_id: str, error: Exception):
        pass

    @abstractmethod
    def to_dict(self) -> dict:
        """
        Returns the configuration needed to recreate the tracing system.
        WARNING: Never share keys or other secret during this process.
        """
        pass

    @classmethod
    @abstractmethod
    def recreate(cls, config: dict) -> "Tracing":
        """
        Recreates a Tracing instance from a configuration dictionary.
        """
        pass

recreate(config) abstractmethod classmethod

Recreates a Tracing instance from a configuration dictionary.

Source code in src/agentswarm/utils/tracing.py

@classmethod
@abstractmethod
def recreate(cls, config: dict) -> "Tracing":
    """
    Recreates a Tracing instance from a configuration dictionary.
    """
    pass

to_dict() abstractmethod

Returns the configuration needed to recreate the tracing system. WARNING: Never share keys or other secret during this process.

Source code in src/agentswarm/utils/tracing.py

@abstractmethod
def to_dict(self) -> dict:
    """
    Returns the configuration needed to recreate the tracing system.
    WARNING: Never share keys or other secret during this process.
    """
    pass

Extensibility

You are not limited to local files. You can implement a Tracing subclass to send data to: - Datadog / New Relic: For enterprise monitoring. - LangSmith / Langfuse: For LLM-specific observability. - Console / Stdout: For real-time streaming to the terminal.

Example: Console Tracer

from agentswarm.utils.tracing import Tracing
from agentswarm.datamodels import Context

class ConsoleTracing(Tracing):
    def trace_agent(self, context: Context, agent_id: str, arguments: dict):
        print(f"🚀 Starting Agent: {agent_id} with inputs: {arguments}")

    def trace_agent_result(self, context: Context, agent_id: str, result: any):
        print(f"✅ Agent {agent_id} completed: {result}")

    # ... implement other methods ...

Local Tracing

The provided LocalTracing implementation writes trace events to JSON files on the local filesystem. This is the default used in examples and allows for post-execution analysis (e.g., using the included trace viewer).

agentswarm.utils.tracing.LocalTracing

Bases: Tracing

Source code in src/agentswarm/utils/tracing.py

class LocalTracing(Tracing):
    def __init__(self, trace_path: str = "./traces"):
        self.trace_path = trace_path

    def trace_agent(self, context: Context, agent_id: str, arguments: dict):
        os.makedirs(self.trace_path, exist_ok=True)
        with open(os.path.join(self.trace_path, f"{context.trace_id}.json"), "a") as f:
            trace_data = {
                "timestamp": datetime.now().isoformat(),
                "type": "agent",
                "step_id": context.step_id,
                "parent_step_id": context.parent_step_id,
                "agent_id": agent_id,
                "arguments": arguments,
                "messages": [message.model_dump() for message in context.messages],
                "store": _get_store_snapshot(context.store),
                "thoughts": context.thoughts,
            }
            f.write(json.dumps(trace_data) + "\n")

    def trace_loop_step(self, context: Context, step_name: str):
        os.makedirs(self.trace_path, exist_ok=True)
        with open(os.path.join(self.trace_path, f"{context.trace_id}.json"), "a") as f:
            trace_data = {
                "timestamp": datetime.now().isoformat(),
                "type": "loop_step",
                "step_id": context.step_id,
                "parent_step_id": context.parent_step_id,
                "agent_id": step_name,  # Use agent_id field to store the step name for UI compatibility
                "arguments": {},
                "messages": [message.model_dump() for message in context.messages],
                "store": _get_store_snapshot(context.store),
                "thoughts": context.thoughts,
            }
            f.write(json.dumps(trace_data) + "\n")

    def trace_agent_result(self, context: Context, agent_id: str, result: Any):
        os.makedirs(self.trace_path, exist_ok=True)
        with open(os.path.join(self.trace_path, f"{context.trace_id}.json"), "a") as f:

            serialized_result = None
            try:
                if hasattr(result, "model_dump"):
                    serialized_result = result.model_dump(mode="json")
                elif hasattr(result, "dict"):
                    serialized_result = result.dict()
                elif isinstance(result, list):
                    serialized_result = []
                    for item in result:
                        if hasattr(item, "model_dump"):
                            serialized_result.append(item.model_dump(mode="json"))
                        elif hasattr(item, "dict"):
                            serialized_result.append(item.dict())
                        else:
                            serialized_result.append(str(item))
                else:
                    serialized_result = str(result)
            except Exception:
                serialized_result = str(result)

            trace_data = {
                "timestamp": datetime.now().isoformat(),
                "type": "agent_result",
                "step_id": context.step_id,
                "parent_step_id": context.parent_step_id,
                "agent_id": agent_id,
                "result": serialized_result,
                "messages": [message.model_dump() for message in context.messages],
                "store": _get_store_snapshot(context.store),
                "thoughts": context.thoughts,
            }
            f.write(json.dumps(trace_data) + "\n")

    def trace_agent_error(self, context: Context, agent_id: str, error: Exception):
        os.makedirs(self.trace_path, exist_ok=True)
        with open(os.path.join(self.trace_path, f"{context.trace_id}.json"), "a") as f:
            trace_data = {
                "timestamp": datetime.now().isoformat(),
                "type": "agent_error",
                "step_id": context.step_id,
                "parent_step_id": context.parent_step_id,
                "agent_id": agent_id,
                "error": str(error),
                "messages": [message.model_dump() for message in context.messages],
                "store": _get_store_snapshot(context.store),
                "thoughts": context.thoughts,
            }
            f.write(json.dumps(trace_data) + "\n")

    def to_dict(self) -> dict:
        from .exceptions import RemoteExecutionNotSupportedError

        raise RemoteExecutionNotSupportedError(
            "LocalTracing cannot be serialized for remote execution."
        )

    @classmethod
    def recreate(cls, config: dict) -> "LocalTracing":
        from .exceptions import RemoteExecutionNotSupportedError

        raise RemoteExecutionNotSupportedError(
            "LocalTracing cannot be recreated from remote configuration."
        )