Composo

Synchronous client for evaluating LLM conversations

Overview

The Composo class provides a synchronous client for evaluating chat messages against custom criteria. Suitable for single evaluations or small batch scenarios with automatic retry mechanisms.

Constructor

from composo import Composo

client = Composo(
    api_key="your_api_key",
    base_url="https://platform.composo.ai",
    num_retries=1,
    model_core=None,
    timeout=60.0
)

Parameters

api_key · string

Your Composo API key for authentication. If not provided, will be loaded from the COMPOSO_API_KEY environment variable.

base_url · string · default: "https://platform.composo.ai"

API base URL. Change only if using a custom Composo deployment.

num_retries · integer · default: "1"

Number of retries on request failure. Each retry uses exponential backoff with jitter. Minimum value is 1 (retries cannot be disabled).

model_core · string

Optional model core identifier for specifying the evaluation model. If not provided, uses the default evaluation model.

timeout · float · default: "60.0"

Request timeout in seconds. Total time to wait for a single request (including retries).

Example

from composo import Composo

# Using API key directly
client = Composo(api_key="your_api_key_here")

# Using environment variable
import os
os.environ["COMPOSO_API_KEY"] = "your_api_key_here"
client = Composo()

# With custom configuration
client = Composo(
    api_key="your_api_key",
    num_retries=3,
    timeout=120.0
)

---

evaluate()

Evaluate messages against one or more evaluation criteria.

result = client.evaluate(
    messages=[...],
    criteria="Your evaluation criterion",
    system=None,
    tools=None,
    result=None,
    block=True,
    tags={"env": "prod"}
)

Parameters

messages · list[dict] · required

List of chat messages to evaluate. Each message should be a dictionary with role and content keys. Mutually exclusive with input.

Supported roles: system, user, assistant, tool

Example:

[
    {"role": "user", "content": "Hello!"},
    {"role": "assistant", "content": "Hi there!"}
]

input · string | list

Raw OpenAI Responses API input — the value passed as input to openai.responses.create(). Mutually exclusive with messages. Use this when evaluating responses from the OpenAI Responses API alongside the result parameter.

Example:

"What are the current prices of Bitcoin and Ethereum in USD?"

criteria · string | list[string]

Evaluation criterion or list of criteria. Can be a custom criterion string or use pre-built criteria from composo.criteria.

Example:

"Reward helpful and accurate responses"
# or
["Criterion 1", "Criterion 2", "Criterion 3"]

system · string

Optional system message to set AI behavior and context for the evaluation.

tools · list[dict]

Optional list of tool definitions for evaluating tool calls. Each tool should follow the OpenAI function calling format.

result · dict | openai.types.responses.Response

Optional LLM result to append to the conversation for evaluation. Accepts a standard dict or an openai.types.responses.Response object returned by openai.responses.create() — Composo auto-detects the type and adapts it automatically.

block · boolean · default: "True"

If False, returns a dictionary with task_id instead of blocking for results. Use for async job submission.

tags · dict[str, Any]

Optional key-value pairs to tag and categorize the request. Tags are useful for organizing, filtering, and analyzing evaluations in analytics tools.

Constraints:

• Keys must be strings, maximum 64 characters
• Values must be strings, numbers, or bools (converted to strings), maximum 64 characters
• No nested structures (dictionaries, lists, tuples, or sets)

Example:

tags={
    "environment": "production",
    "version": "1.0.0",
    "experiment": "variant_a"
}

evaluate_latest · boolean

Whether to evaluate only the latest assistant response (True) or all assistant responses (False). If not provided, defaults to True for chat evaluations.

Note: Lightning model cores (align-lightning-*) only support True.

data_retention · string · default: "standard"

Requires composo 0.4.0 or later.

How long Composo may keep the content of this request. "standard" stores it as normal. "none" evaluates and returns the score without storing anything but the usage record: no trace, no evaluation record, and a redacted request payload.

Note: requests sent with "none" do not appear in Insights — there is no stored evaluation for them to aggregate.

Resolved against your account-level retention setting by most-restrictive-wins, so this can only ever tighten retention for a request, never loosen it.

explanation_cleaning · string

When set to "end_user", the response will include a cleaned_explanation field that rewrites the explanation to only reference content visible in user and assistant messages.

Returns

result · EvaluationResponse | list[EvaluationResponse]

• Returns single EvaluationResponse if one criterion provided
• Returns list[EvaluationResponse] if multiple criteria provided
• Returns dict with task_id if block=False

Response Schema

EvaluationResponse

score · float | null

Evaluation score between 0.0 and 1.0. Returns null if the criterion was deemed not applicable.

explanation · string

Detailed explanation of the evaluation score and reasoning.

cleaned_explanation · string | null

A rewrite of explanation that only references content visible in user and assistant messages. Only present when explanation_cleaning="end_user" is set in the request.

Examples

Basic Evaluation

from composo import Composo

client = Composo()

messages = [
    {"role": "user", "content": "What's the capital of France?"},
    {"role": "assistant", "content": "The capital of France is Paris."}
]

result = client.evaluate(
    messages=messages,
    criteria="Reward accurate and informative responses",
    tags={"environment": "production", "version": "1.0.0"}
)

print(f"Score: {result.score}")
# Output: Score: 0.95

print(f"Explanation: {result.explanation}")
# Output: Explanation: The response correctly identifies Paris as the capital of France...

Multiple Criteria Evaluation

results = client.evaluate(
    messages=[...],
    criteria=[
        "Reward accurate information",
        "Reward clear communication",
        "Penalize overly technical jargon"
    ]
)

for result in results:
    print(f"Score: {result.score} - {result.explanation}")

Tool Call Evaluation

messages = [
    {"role": "user", "content": "What's the weather in SF?"},
    {
        "role": "assistant",
        "content": None,
        "tool_calls": [{
            "id": "call_123",
            "type": "function",
            "function": {
                "name": "get_weather",
                "arguments": '{"location": "San Francisco"}'
            }
        }]
    },
    {
        "role": "tool",
        "tool_call_id": "call_123",
        "content": '{"temp": 65, "condition": "sunny"}'
    },
    {"role": "assistant", "content": "It's 65°F and sunny in San Francisco!"}
]

tools = [{
    "type": "function",
    "function": {
        "name": "get_weather",
        "description": "Get current weather for a location",
        "parameters": {
            "type": "object",
            "properties": {
                "location": {"type": "string"}
            }
        }
    }
}]

result = client.evaluate(
    messages=messages,
    tools=tools,
    criteria="Reward correct tool usage and accurate responses"
)

Non-blocking Evaluation

# Submit evaluation without waiting
response = client.evaluate(
    messages=[...],
    criteria="Your criterion",
    block=False
)

task_id = response["task_id"]
print(f"Task submitted with ID: {task_id}")
# Use task_id to check status later

OpenAI Responses API — Built-in Tools

Pass the Response object returned by openai.responses.create() directly as result. Use input instead of messages to match the Responses API’s input format.

Known limitation: Composo evaluates only the context provided in the current call — it cannot follow the previous_response_id chain to reconstruct prior turns. If your workflow uses multi-turn Responses API conversations (i.e. passing previous_response_id to link responses), make sure to pass the full conversation history explicitly via messages rather than relying on input + result alone.

from openai import OpenAI
from composo import Composo

openai_client = OpenAI()
composo_client = Composo()

input_text = (
    "Search the web for the latest news story on SpaceX and write a Python script "
    "to plot their latest 3 launch dates"
)

response = openai_client.responses.create(
    model="gpt-4.1",
    input=input_text,
    tools=[
        {"type": "web_search"},
        {"type": "code_interpreter", "container": {"type": "auto"}},
    ],
)

result = composo_client.evaluate(
    input=input_text,
    result=response,
    criteria="Reward responses that search the web and produce Python code based on the results",
)

print(f"Score: {result.score}")
print(f"Explanation: {result.explanation}")

OpenAI Responses API — Remote MCP Server

MCP_TOOL = {
    "type": "mcp",
    "server_label": "coingecko",
    "server_url": "https://mcp.api.coingecko.com/mcp",
    "require_approval": "never",
}

input_text = "What are the current prices of Bitcoin and Ethereum in USD?"

response = openai_client.responses.create(
    model="gpt-4.1",
    input=input_text,
    tools=[MCP_TOOL],
)

result = composo_client.evaluate(
    input=input_text,
    result=response,
    criteria="Reward responses that use the MCP tool to retrieve live cryptocurrency prices and present them clearly",
)

print(f"Score: {result.score}")
print(f"Explanation: {result.explanation}")

---

evaluate_trace()

Evaluate multi-agent traces with full conversation history across multiple agents.

result = client.evaluate_trace(
    trace=trace_object,
    criteria="Your evaluation criterion",
    model_core=None,
    block=True,
    tags={"env": "prod"}
)

Parameters

trace · MultiAgentTrace · required

Multi-agent trace object containing agent interactions, initial input, and final output.

criteria · string | list[string] · required

Evaluation criterion or list of criteria for trace evaluation.

model_core · ModelCore

Optional model core identifier for trace evaluation.

block · boolean · default: "True"

If False, returns a dictionary with task_id instead of blocking for results.

tags · dict[str, Any]

Optional key-value pairs to tag and categorize the request. Tags are useful for organizing, filtering, and analyzing trace evaluations in analytics tools.

Constraints:

• Keys must be strings, maximum 64 characters
• Values must be strings, numbers, or bools (converted to strings), maximum 64 characters
• No nested structures (dictionaries, lists, tuples, or sets)

Example:

tags={
    "environment": "production",
    "agent_version": "2.1.0",
    "experiment": "improved_prompts"
}

evaluate_latest · boolean

Whether to evaluate only the latest response (True) or all responses (False). If not provided, defaults to False for trace evaluations.

Note: Must be False for trace evaluations.

data_retention · string · default: "standard"

Requires composo 0.4.0 or later.

How long Composo may keep the content of this trace. "standard" stores it as normal. "none" evaluates and returns the score without storing the trace, the evaluation record, or an unredacted request payload — such requests do not appear in Insights.

Resolved against your account-level retention setting by most-restrictive-wins.

Returns

result · MultiAgentTraceResponse | list[MultiAgentTraceResponse]

• Returns single MultiAgentTraceResponse if one criterion provided
• Returns list[MultiAgentTraceResponse] if multiple criteria provided
• Returns dict with task_id if block=False

Response Schema

MultiAgentTraceResponse

agent_scores · dict

Per-agent evaluation scores mapping agent IDs to their individual scores.

overall_score · float

Overall trace score aggregated across all agents.

explanation · string

Detailed explanation of the trace evaluation.

criterion · string

The criterion that was evaluated.

Example

from composo import Composo, ComposoTracer, Instruments, AgentTracer
from openai import OpenAI

# Initialize tracing
ComposoTracer.init(instruments=Instruments.OPENAI)
openai_client = OpenAI()
composo_client = Composo()

# Use AgentTracer context manager to capture trace
with AgentTracer(name="research_agent") as tracer:
    response = openai_client.chat.completions.create(
        model="gpt-4",
        messages=[{"role": "user", "content": "Research: quantum computing"}]
    )
    result = response.choices[0].message.content

    # Get the trace object
    trace = tracer.trace

# Evaluate the captured trace
evaluation = composo_client.evaluate_trace(
    trace=trace,
    criteria="Reward thorough research and accurate information",
    tags={"environment": "production", "agent_version": "1.0.0"}
)

print(f"Overall Score: {evaluation.overall_score}")
print(f"Explanation: {evaluation.explanation}")

---

Context Manager Usage

The Composo client supports context managers for automatic resource cleanup:

with Composo() as client:
    result = client.evaluate(
        messages=[...],
        criteria="Your criterion"
    )
    print(result.score)
# Client automatically closed