responsible-ai-platform 1.0.1

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 — LangGraph, LangChain, CrewAI, 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.

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_AGENT_VERSION=1.0.0
RAIA_MODEL_VERSION=claude-sonnet-4-6
RAIA_ENVIRONMENT=dev
RAIA_MAX_STEPS_ALLOWED=15
RAIA_DEBUG=true

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..."
TOOL_REGISTRY = ["search_products", "get_order_details"]

# Configure once at startup (optional — sets defaults for all traces)
AgentTrace.configure(
    system_prompt=SYSTEM_PROMPT,
    tool_registry=TOOL_REGISTRY,
)

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.

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.

from responsible_ai_platform.agentic import trace, tool

@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.

Option 4: Manual Logging (Any Framework)

For full control over what gets logged.

from responsible_ai_platform.agentic import AgentTrace

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}
        ],
        tool_results=[
            {"name": "search_products", "result": "Found 5 laptops..."}
        ],
    )
    trace.set_outcome("success")

API Reference

AgentTrace

The core tracing class.

Class Method

AgentTrace.configure(
    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
    tool_registry=None,    # List of authorized tool names
)

Constructor

trace = AgentTrace(
    app_id=None,               # Agent application ID
    task_description="",       # What this trace is about
    session_id=None,           # Session ID (auto-generated if omitted)
    max_steps_allowed=None,    # Max tool steps
    metadata=None,             # Extra metadata dict
    system_prompt=None,        # System prompt text
    tool_registry=None,        # List of authorized tool names
    expected_outcome=None,     # Ground truth for evaluation
)

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.
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
)

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
)

Auto-extracts from messages:

• Input/Output: first HumanMessage and last AIMessage
• Tool calls: from AIMessage.tool_calls with is_authorized=True
• 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

Trace Output Format

Each trace is a JSON array of entries (one per user interaction):

[
  {
    "trace_id": "uuid",
    "session_id": "uuid",
    "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}
    ],
    "tool_results": [
      {"name": "search_products", "result": "Found 5 laptops..."}
    ],
    "tool_registry": ["search_products", "get_order_details"],
    "agent_thinking": [
      {"step": 1, "thought": "User wants laptop recommendations", "action": "search_products"}
    ],
    "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