responsible-ai-platform 2.0.0

Responsible AI Platform — SDK for tracing, evaluation, and governance of AI assets

Responsible AI Platform SDK

Python SDK for sending agent trace data to the RAIA — Responsible AI Assessment platform. Works with any agent framework — LangChain, LangGraph, LangChain-AWS (Bedrock), CrewAI, AutoGen, or custom Python agents.

Trace logs are uploaded as JSON to the RAIA backend and used by the evaluation service to compute metrics like latency, token usage, tool governance, safety, quality, and readiness — at both the system level and per agent in a multi-agent system.

Every asset declares an agent topology: a single-agent system has one orchestrator node, a multi-agent system adds sub-agents under it. See Agent Topology (required) below.

Installation

pip install responsible-ai-platform

Requires Python ≥ 3.9.

Configuration

API Key (Required)

Generate an API key in the RAIA UI (Discover → your asset → Generate API Key). Then add to your .env:

RAIA_API_KEY=raia_...
RAIA_API_BASE_URL=https://your-raia-host.example.com

That's it — just two variables. The API key is a random opaque token; the server looks up your asset context by hashing it, so no project/tenant info needs to live in the key itself.

Variable	Required	Description
RAIA_API_KEY	Yes	API key from the RAIA UI (valid 90 days — regenerate when it expires)
RAIA_API_BASE_URL	Yes	RAIA backend URL

Optional Variables

RAIA_APP_ID=my-agent-app
RAIA_AGENT_VERSION=1.0.0
RAIA_MODEL_VERSION=claude-sonnet-4-6
RAIA_ENVIRONMENT=dev
RAIA_DEBUG=true

Per-agent step caps (max_steps_allowed, optimal_steps) are declared on each topology node — see Agent Topology below.

Agent Topology (required)

Every asset is modeled as a topology — a list of one or more agent nodes. There is no "single-agent default" or "multi-agent toggle"; the data model is uniform.

• A single-agent system declares one node with role="orchestrator".
• A multi-agent system declares an orchestrator plus sub-agents under it.

Call AgentTrace.configure(agents=[…]) once at startup. The SDK raises if you try to open a trace without it. The agent names you declare here must match the names registered in the RAIA UI Configuration page — that's the join key the backend uses for per-agent metric attribution.

What you get when the topology has more than one agent:

• per-agent metric breakdowns (operational, quality, safety) alongside the aggregate;
• per-agent tool governance — a sub-agent's tool isn't flagged as hallucinated just because it isn't in the orchestrator's registry;
• per-trace agent attribution so the Traces tab shows which agent emitted each entry and which agent invoked each tool call.

Declaring the topology

Single-agent system — declare one orchestrator node:

from responsible_ai_platform.agentic import AgentTrace

AgentTrace.configure(
    app_id="movie-bot",
    system_prompt="...",
    agents=[
        {
            "agent_name": "movie-bot",
            "role": "orchestrator",
            "tools": ["search_movies", "get_showtimes"],
            "model_version": "claude-sonnet-4-6",
        },
    ],
)

Multi-agent system — orchestrator + sub-agents:

AgentTrace.configure(
    app_id="movie-bot",
    system_prompt="...",
    agents=[
        {
            "agent_name": "orchestrator",
            "role": "orchestrator",
            "tools": ["plan", "delegate"],
            "model_version": "claude-sonnet-4-6",
        },
        {
            "agent_name": "researcher",
            "role": "sub_agent",
            "parent_agent_name": "orchestrator",
            "tools": ["web_search", "fetch_url"],
        },
        {
            "agent_name": "summarizer",
            "role": "sub_agent",
            "parent_agent_name": "orchestrator",
            "tools": [],
        },
    ],
)

Each topology node accepts: agent_name (required), role (orchestrator | sub_agent | worker | router | tool_runner), parent_agent_name (None for the root), tools / tool_registry (list of authorized tool names for this agent), model_version, framework, max_steps_allowed, optimal_steps, agent_actions, boundary_definitions, escalation_conditions, ground_truth_description. Exactly one node must have role="orchestrator".

@trace_subagent("name") will auto-register the sub-agent on the topology if you didn't pre-declare it via configure(), but the orchestrator node is always required.

Minimal single-agent quick start

from responsible_ai_platform.agentic import AgentTrace

AgentTrace.configure(
    agents=[
        {"agent_name": "my-bot", "role": "orchestrator", "tools": ["search", "fetch"]},
    ],
)

with AgentTrace(task_description="answer a question") as t:
    t.log_interaction(
        input_text="What's the capital of France?",
        output_text="Paris.",
        prompt_tokens=12, completion_tokens=2, total_tokens=14,
        success=True,
    )
    t.set_outcome("success")
# Trace uploaded automatically.

@trace_subagent decorator (recommended for custom multi-agent code)

Wrap any function with @trace_subagent("name") to scope the active sub-agent for the duration of the call. Nested @tool invocations and any log_interaction() calls inside automatically get tagged with that agent.

from responsible_ai_platform.agentic import trace, trace_subagent, tool

@tool
def web_search(q): ...

@tool
def summarize(text): ...

@trace_subagent("researcher", parent="orchestrator")
def do_research(query):
    return web_search(query)              # tagged invoked_by_agent_name="researcher"

@trace_subagent("summarizer", parent="orchestrator")
def do_summary(text):
    return summarize(text)                # tagged invoked_by_agent_name="summarizer"

@trace(task_description="Movie research")
def run_agent(q):
    raw = do_research(q)
    return do_summary(raw)

LangGraph supervisor / swarm graphs

LangGraph multi-agent graphs (e.g. create_supervisor, swarm patterns) tag each AIMessage with the originating node name. The SDK reads that automatically. For richer per-agent metrics, use log_langgraph_steps_per_agent instead of the single-entry helper:

from responsible_ai_platform.agentic import AgentTrace
from responsible_ai_platform.agentic.integrations import log_langgraph_steps_per_agent

AgentTrace.configure(
    agents=[
        {"agent_name": "supervisor", "role": "orchestrator", "tools": []},
        {"agent_name": "researcher", "role": "sub_agent",
         "parent_agent_name": "supervisor", "tools": ["search"]},
        {"agent_name": "writer", "role": "sub_agent",
         "parent_agent_name": "supervisor", "tools": []},
    ],
)

with AgentTrace(task_description="...") as trace:
    result = graph.invoke({"messages": [HumanMessage(content=query)]})
    log_langgraph_steps_per_agent(trace, result["messages"], user_input=query)
    trace.set_outcome("success")

log_langgraph_steps_per_agent emits one entry per LangGraph node that participated in the run, so the backend can compute per-sub-agent operational and quality metrics. The plain log_langgraph_steps still works — it produces one entry summarizing the whole invocation but each tool call is still tagged with the originating node so per-agent tool governance still works.

Manual per-call tagging

If you don't use the decorators, pass agent_name directly:

trace.log_interaction(
    input_text=query,
    output_text=response,
    agent_name="researcher",          # Tags this entry as the researcher
    tool_calls=[
        {"name": "web_search", "arguments": {"q": query},
         "is_authorized": True,
         "invoked_by_agent_name": "researcher"},
    ],
)

Handoffs between agents

Record explicit handoffs (orchestrator → sub-agent transfers) for the multi-agent metrics:

trace.log_handoff(
    from_agent="orchestrator",
    to_agent="researcher",
    reason="user asked about facts; needs lookup",
)

Integration Options

Option 1: LangGraph / LangChain (Recommended)

Use the built-in integration that auto-extracts tool calls, token usage, input/output, and thinking steps from LangGraph messages.

from datetime import datetime, timezone
from langchain_core.messages import HumanMessage, SystemMessage
from langgraph.prebuilt import create_react_agent

from responsible_ai_platform.agentic import AgentTrace
from responsible_ai_platform.agentic.integrations import log_langgraph_steps

SYSTEM_PROMPT = "You are a helpful assistant..."

# Configure once at startup (required — every asset declares its topology).
AgentTrace.configure(
    system_prompt=SYSTEM_PROMPT,
    agents=[
        {
            "agent_name": "support-bot",
            "role": "orchestrator",
            "tools": ["search_products", "get_order_details"],
        },
    ],
)

agent = create_react_agent(llm, tools, prompt=SystemMessage(content=SYSTEM_PROMPT))

# Per-call usage (one trace per request)
with AgentTrace(task_description="Customer support query") as trace:
    start_time = datetime.now(timezone.utc)
    result = agent.invoke({"messages": [HumanMessage(content="Show me laptops")]})
    end_time = datetime.now(timezone.utc)

    log_langgraph_steps(
        trace,
        result["messages"],
        user_input="Show me laptops",
        start_time=start_time,
        end_time=end_time,
    )
    trace.set_outcome("success")
# Trace auto-uploads to S3 on exit

Option 2: Session-Based Tracing (Multi-Turn Chat)

For chat applications where one conversation = one trace file with multiple entries. AgentTrace.configure(agents=[…]) must already have been called at startup.

from responsible_ai_platform.agentic import AgentTrace
from responsible_ai_platform.agentic.integrations import log_langgraph_steps

trace = AgentTrace(
    task_description="Customer support session",
    session_id="unique-session-id",
)
trace._async_upload = False   # Synchronous uploads (reliable)
trace._auto_upload = True     # Auto-upload after each message
trace.start()

def handle_message(user_input: str):
    start_time = datetime.now(timezone.utc)
    result = agent.invoke({"messages": [HumanMessage(content=user_input)]})
    end_time = datetime.now(timezone.utc)

    log_langgraph_steps(
        trace,
        result["messages"],
        user_input=user_input,
        start_time=start_time,
        end_time=end_time,
    )
    # With _auto_upload=True, the trace JSON is uploaded after each call.

# When the session ends
trace.set_outcome("success")
trace.finish()  # Final upload

Option 3: Decorator-Based (Custom Agents)

For custom Python agents without a framework. Use @trace on the agent entry point and @tool on tool functions. AgentTrace.configure(agents=[…]) must be called once at startup so the trace decorator knows which agent to attribute the entries to.

from responsible_ai_platform.agentic import AgentTrace, trace, tool

# 1. Declare topology once at startup
AgentTrace.configure(
    agents=[
        {
            "agent_name": "support-bot",
            "role": "orchestrator",
            "tools": ["search_products", "get_order"],
        },
    ],
)

@tool
def search_products(query: str) -> str:
    """Search the product catalog."""
    return results

@tool
def get_order(order_id: str) -> dict:
    """Look up an order."""
    return {"order_id": order_id, "status": "delivered"}

@trace(task_description="Handle customer query")
def my_agent(query: str) -> str:
    results = search_products(query)
    return f"Found: {results}"

my_agent("Show me laptops under $500")
# Trace auto-created, all @tool calls logged, uploaded to S3.

The @tool decorator auto-captures: tool name, arguments, result, latency, and errors. It requires an active @trace context — if no trace is active, the function runs normally. For multi-agent code, wrap sub-agent functions with @trace_subagent("name") and the nested @tool calls will be tagged with that sub-agent.

Option 4: Manual Logging (Any Framework)

For full control over what gets logged. As with every other path, AgentTrace.configure(agents=[…]) must be called once before opening a trace.

from responsible_ai_platform.agentic import AgentTrace

AgentTrace.configure(
    agents=[
        {
            "agent_name": "support-bot",
            "role": "orchestrator",
            "tools": ["search_products"],
        },
    ],
)

with AgentTrace(task_description="My agent task") as trace:
    trace.log_interaction(
        input_text="What laptops do you have?",
        output_text="Here are our top laptops...",
        model="claude-sonnet-4-6",
        prompt_tokens=150,
        completion_tokens=200,
        total_tokens=350,
        success=True,
        tool_calls=[
            {
                "name": "search_products",
                "arguments": {"query": "laptops"},
                "is_authorized": True,
                # invoked_by_agent_name is auto-stamped from the entry's
                # agent_name; pass it explicitly only when the same entry
                # contains tool calls from multiple sub-agents.
            }
        ],
        tool_results=[
            {"name": "search_products", "result": "Found 5 laptops..."}
        ],
    )
    trace.set_outcome("success")

API Reference

AgentTrace

The core tracing class.

Class Method

AgentTrace.configure(
    agents,                # Required — list of agent topology nodes
    tenant_id=None,        # Override tenant from .env
    app_id=None,           # Override app_id from .env
    agent_version=None,    # Override agent_version from .env
    model_version=None,    # Override model_version from .env
    environment=None,      # Override environment from .env
    system_prompt=None,    # Default system prompt for all traces
)

Constructor

trace = AgentTrace(
    task_description="",       # What this trace is about
    session_id=None,           # Session ID (auto-generated if omitted)
    metadata=None,             # Extra metadata dict
    system_prompt=None,        # System prompt text
    expected_outcome=None,     # Ground truth for evaluation
    app_id=None,               # Override app_id from .env / configure()
    agents=None,               # Per-trace topology override (rare — usually
                               # configure() once at startup)
)

The class-level topology (set via configure()) is used by default. Per-agent settings like max_steps_allowed, optimal_steps, tool_registry, escalation_conditions all live on individual nodes inside agents=[…].

Methods

Method	Description
start()	Start the trace timer. Called automatically when using with.
finish()	Finalize and upload the trace. Called automatically when using with.
log_interaction(...)	Log a single user-agent interaction (message pair).
log_step(...)	Log a single tool invocation (used by @tool decorator).
set_outcome(outcome, escalation_reason=None)	Set outcome: "success", "failure", "partial", "escalated".
log_boundary_violation(action, rule_violated)	Record a policy constraint violation.
log_handoff(from_agent, to_agent, reason=None)	Record an agent-to-agent handoff. Surfaces in the multi-agent metrics.
to_dict()	Serialize trace as list of entry dicts.

log_interaction() Parameters

trace.log_interaction(
    input_text="user query",           # User message
    output_text="agent response",      # Agent response
    start_time=None,                   # datetime (defaults to now)
    end_time=None,                     # datetime (defaults to now)
    model=None,                        # LLM model used
    prompt_tokens=0,                   # Input token count
    completion_tokens=0,               # Output token count
    total_tokens=0,                    # Total tokens
    success=True,                      # Whether interaction succeeded
    error_type=None,                   # Exception class name
    error_message=None,                # Error details
    tool_calls=None,                   # List of {name, arguments, is_authorized}
    tool_results=None,                 # List of {name, result}
    agent_thinking=None,               # List of reasoning steps
    num_steps=None,                    # Number of agent steps
    task_description=None,             # Per-interaction task description
    system_prompt=None,                # System prompt override
    expected_outcome=None,             # Ground truth override
    boundary_violations=None,          # List of violations
    escalation_events=None,            # List of escalation events
    escalation_reason=None,            # Escalation reason text
    agent_name=None,                   # Which agent emitted this entry. Defaults
                                       # to the current @trace_subagent context,
                                       # or to the orchestrator declared in
                                       # configure(agents=…).
    step_index=None,                   # Optional step number within the trace
)

@trace_subagent

Decorator that scopes the active sub-agent for a function call. Any @tool invocations or log_interaction() calls inside are auto-tagged.

@trace_subagent(
    name="researcher",          # Stable name — must match the topology / Configuration UI
    parent="orchestrator",      # Parent agent (defaults to the orchestrator)
    role="sub_agent",           # Logical role
)
def do_research(query): ...

If the named sub-agent isn't already in the topology declared via AgentTrace.configure(agents=...), it's auto-registered the first time the decorator runs.

log_langgraph_steps()

One-line integration for LangGraph/LangChain agents.

from responsible_ai_platform.agentic.integrations import log_langgraph_steps

log_langgraph_steps(
    trace,                  # Active AgentTrace instance
    messages,               # List of LangChain message objects from agent.invoke()
    user_input=None,        # Original user query (auto-detected if omitted)
    start_time=None,        # When the invocation started
    end_time=None,          # When the invocation ended
    agent_name=None,        # Explicit override for the entry's agent. When
                            # omitted, the SDK uses the LangGraph node name
                            # off the AI message, falling back to the
                            # orchestrator declared in configure().
)

Auto-extracts from messages:

• Input/Output: first HumanMessage and last AIMessage
• Tool calls: from AIMessage.tool_calls with is_authorized=True. Each call carries invoked_by_agent_name taken from the LangGraph node that emitted the AI message (msg.name / response_metadata.langgraph_node / additional_kwargs.langgraph_node).
• Tool results: from ToolMessage objects, paired by tool_call_id
• Token usage: from AIMessage.usage_metadata (input_tokens, output_tokens)
• System prompt: from SystemMessage if present
• Model: from AIMessage.response_metadata
• Thinking steps: reconstructed from AI message + tool call sequence

log_langgraph_steps_per_agent()

Multi-agent variant: emit one entry per LangGraph node that participated in the run. Use this for supervisor / swarm graphs when you want richer per-sub-agent operational and quality metrics.

from responsible_ai_platform.agentic.integrations import log_langgraph_steps_per_agent

log_langgraph_steps_per_agent(
    trace,
    messages,
    user_input=None,
    start_time=None,
    end_time=None,
)

For single-node graphs this falls back to the same behavior as log_langgraph_steps.

Trace Output Format

Each trace is a JSON array of entries (one per user interaction). The schema version is "2.0". Every entry carries the agent attribution fields (agent_name, parent_agent_name, agent_path, topology, tool_registry_by_agent, handoffs, and invoked_by_agent_name on each tool call) — single-agent runs use the orchestrator name you declared in configure(agents=[…]), multi-agent runs use the per-call sub-agent name.

[
  {
    "schema_version": "2.0",
    "trace_id": "uuid",
    "session_id": "uuid",
    "span_id": "uuid",
    "parent_span_id": "uuid",

    "agent_name": "researcher",
    "parent_agent_name": "orchestrator",
    "agent_path": ["orchestrator", "researcher"],
    "step_index": 1,

    "start_time": "2026-04-20T10:15:30+00:00",
    "end_time": "2026-04-20T10:15:31+00:00",
    "latency": 1000.0,
    "input": "Show me laptops",
    "output": "Here are our top laptops...",
    "system_prompt": "You are a helpful assistant...",
    "task_description": "Customer support query",
    "expected_outcome": null,
    "model": "claude-sonnet-4-6",
    "prompt_tokens": 150,
    "completion_tokens": 200,
    "total_tokens": 350,
    "success": true,
    "status": "success",
    "error_type": null,
    "error_message": null,

    "tool_calls": [
      {
        "name": "search_products",
        "arguments": {"query": "laptops"},
        "is_authorized": true,
        "invoked_by_agent_name": "researcher"
      }
    ],
    "tool_results": [
      {"name": "search_products", "result": "Found 5 laptops..."}
    ],
    "tool_registry_by_agent": {
      "orchestrator": ["plan", "delegate"],
      "researcher": ["search_products", "get_order_details"]
    },

    "topology": [
      {"agent_name": "orchestrator", "role": "orchestrator",
       "parent_agent_name": null, "tools": ["plan", "delegate"]},
      {"agent_name": "researcher", "role": "sub_agent",
       "parent_agent_name": "orchestrator",
       "tools": ["search_products", "get_order_details"]}
    ],
    "handoffs": [
      {"from_agent": "orchestrator", "to_agent": "researcher",
       "reason": "needs catalog lookup",
       "timestamp": "2026-04-20T10:15:30+00:00"}
    ],

    "agent_thinking": [
      {"step": 1, "thought": "User wants laptop recommendations",
       "action": "search_products", "agent_name": "researcher"}
    ],
    "num_steps": 1,
    "boundary_violations": [],
    "escalation_events": [],
    "escalation_reason": null,
    "app_id": "my-agent-app",
    "tenant_id": "MyTenant",
    "agent_version": "1.0.0",
    "environment": "dev"
  }
]

S3 Upload Path

Traces are uploaded to:

{tenant_name}/{analysis_type}/{project_name}/files/{session_id}.json

• With _auto_upload=True, the file is overwritten after each message (growing array).
• On finish(), a final upload is done with the complete trace.

Error Handling & Limits

• Upload failure — if the RAIA API is unreachable or returns an error, the trace is dropped with a warning log. There's no on-disk retry queue.
• Upload size cap — requests over 50 MB are refused locally (prevents runaway payloads).
• Upload queue cap — up to 100 in-flight async uploads at a time. Beyond that, new traces are dropped with a warning. Increase concurrency in your process if you need more.
• Authentication errors — raised immediately so you can fix RAIA_API_KEY / RAIA_API_BASE_URL.
• Per-interaction errors — captured inside the trace (success=false, error_type, error_message). The trace itself still uploads.

License

Apache 2.0. Copyright 2026 Cirrus Labs — RAIA.

Issues & Contributions

• Issues: https://github.com/CL-AI-COE/trace-sdk/issues
• Repository: https://github.com/CL-AI-COE/trace-sdk