> ## Documentation Index
> Fetch the complete documentation index at: https://docs.composo.ai/llms.txt
> Use this file to discover all available pages before exploring further.

# AsyncComposo

> Asynchronous client for high-performance batch evaluations

## Overview

The `AsyncComposo` class provides an asynchronous client for evaluating chat messages with support for concurrent processing. Ideal for large batch evaluation scenarios and high-throughput applications.

## Constructor

```python theme={null}
from composo import AsyncComposo

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

### Parameters

<ParamField path="api_key" type="string" optional>
  Your Composo API key for authentication. If not provided, will be loaded from the `COMPOSO_API_KEY` environment variable.
</ParamField>

<ParamField path="base_url" type="string" default="https://platform.composo.ai">
  API base URL. Change only if using a custom Composo deployment.
</ParamField>

<ParamField path="num_retries" type="integer" default="1">
  Number of retries on request failure. Each retry uses exponential backoff with jitter. Minimum value is 1 (retries cannot be disabled).
</ParamField>

<ParamField path="model_core" type="string" optional>
  Optional model core identifier for specifying the evaluation model.
</ParamField>

<ParamField path="max_concurrent_requests" type="integer" default="5">
  Maximum number of concurrent API requests. Controls throughput and prevents rate limit issues.

  **Recommendations:**

  * `5-10`: Most use cases
  * `20+`: High-performance scenarios with adequate rate limits
</ParamField>

<ParamField path="timeout" type="float" default="60.0">
  Request timeout in seconds. Total time to wait for a single request (including retries).
</ParamField>

### Example

```python theme={null}
from composo import AsyncComposo
import asyncio

async def main():
    # Using API key directly
    client = AsyncComposo(api_key="your_api_key_here")

    # With custom concurrency
    client = AsyncComposo(
        api_key="your_api_key",
        max_concurrent_requests=10,
        num_retries=3
    )

asyncio.run(main())
```

***

## evaluate()

Asynchronously evaluate messages against one or more evaluation criteria.

```python theme={null}
result = await client.evaluate(
    messages=[...],
    criteria="Your evaluation criterion",
    system=None,
    tools=None,
    result=None,
    block=True,
    tags=None
)
```

### Parameters

<ParamField path="messages" type="list[dict]" required>
  List of chat messages to evaluate. Each message should be a dictionary with `role` and `content` keys.

  **Supported roles:** `system`, `user`, `assistant`, `tool`
</ParamField>

<ParamField path="criteria" type="string | list[string]" optional>
  Evaluation criterion or list of criteria. Multiple criteria are evaluated concurrently for better performance.
</ParamField>

<ParamField path="system" type="string" optional>
  Optional system message to set AI behavior and context.
</ParamField>

<ParamField path="tools" type="list[dict]" optional>
  Optional list of tool definitions for evaluating tool calls.
</ParamField>

<ParamField path="result" type="dict" optional>
  Optional LLM result to append to the conversation.
</ParamField>

<ParamField path="block" type="boolean" default="True">
  If `False`, returns a dictionary with `task_id` instead of blocking for results.
</ParamField>

<ParamField path="tags" type="dict[str, Any]" optional>
  Optional key-value pairs to tag and categorize the request. Tags are useful for organizing, filtering, and analyzing evaluations in Metabase or other analytics tools.

  **Constraints:**

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

  **Example:**

  ```python theme={null}
  tags={
      "environment": "production",
      "version": "1.0.0",
      "experiment": "variant_a"
  }
  ```
</ParamField>

<ParamField path="evaluate_latest" type="boolean" optional>
  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`.
</ParamField>

<ParamField path="explanation_cleaning" type="string" optional>
  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.
</ParamField>

### Returns

<ResponseField name="result" type="EvaluationResponse | list[EvaluationResponse]">
  * Returns single `EvaluationResponse` if one criterion provided
  * Returns `list[EvaluationResponse]` if multiple criteria provided (evaluated concurrently)
  * Returns `dict` with `task_id` if `block=False`
</ResponseField>

### Response Schema

**EvaluationResponse**

<ResponseField name="score" type="float | null">
  Evaluation score between 0.0 and 1.0. Returns `null` if criterion not applicable.
</ResponseField>

<ResponseField name="explanation" type="string">
  Detailed explanation of the evaluation score.
</ResponseField>

<ResponseField name="cleaned_explanation" type="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.
</ResponseField>

### Examples

#### Basic Async Evaluation

```python theme={null}
from composo import AsyncComposo
import asyncio

async def evaluate_single():
    async with AsyncComposo() as client:
        messages = [
            {"role": "user", "content": "What's 2+2?"},
            {"role": "assistant", "content": "2+2 equals 4."}
        ]

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

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

asyncio.run(evaluate_single())
```

#### Batch Evaluation with Concurrency

```python theme={null}
from composo import AsyncComposo
import asyncio

async def batch_evaluate():
    async with AsyncComposo(max_concurrent_requests=10) as client:
        # Prepare multiple evaluations
        conversations = [
            [{"role": "user", "content": "Hello"}],
            [{"role": "user", "content": "Goodbye"}],
            [{"role": "user", "content": "Help me"}],
            # ... more conversations
        ]

        # Create tasks for concurrent evaluation
        tasks = [
            client.evaluate(
                messages=conv,
                criteria="Reward helpful responses"
            )
            for conv in conversations
        ]

        # Execute all evaluations concurrently
        results = await asyncio.gather(*tasks)

        for i, result in enumerate(results):
            print(f"Conversation {i}: Score = {result.score}")

asyncio.run(batch_evaluate())
```

#### Multiple Criteria (Evaluated Concurrently)

```python theme={null}
async def evaluate_multi_criteria():
    async with AsyncComposo() as client:
        result = await client.evaluate(
            messages=[...],
            criteria=[
                "Reward accurate information",
                "Reward clear communication",
                "Penalize inappropriate tone"
            ]
        )

        # All criteria evaluated concurrently
        for res in result:
            print(f"Score: {res.score}")

asyncio.run(evaluate_multi_criteria())
```

#### High-Performance Batch Processing

```python theme={null}
from composo import AsyncComposo
import asyncio

async def process_large_dataset():
    # Configure for high throughput
    async with AsyncComposo(max_concurrent_requests=20) as client:
        # Process 1000 conversations
        conversations = load_conversations()  # Your data loading function

        # Split into batches to avoid memory issues
        batch_size = 100
        all_results = []

        for i in range(0, len(conversations), batch_size):
            batch = conversations[i:i+batch_size]

            tasks = [
                client.evaluate(
                    messages=conv,
                    criteria="Your criterion"
                )
                for conv in batch
            ]

            batch_results = await asyncio.gather(*tasks)
            all_results.extend(batch_results)

            print(f"Processed {len(all_results)} / {len(conversations)}")

        return all_results

asyncio.run(process_large_dataset())
```

***

## evaluate\_trace()

Asynchronously evaluate multi-agent traces.

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

### Parameters

<ParamField path="trace" type="MultiAgentTrace" required>
  Multi-agent trace object containing agent interactions.
</ParamField>

<ParamField path="criteria" type="string | list[string]" required>
  Evaluation criterion or list of criteria. Multiple criteria are evaluated concurrently.
</ParamField>

<ParamField path="model_core" type="ModelCore" optional>
  Optional model core identifier.
</ParamField>

<ParamField path="block" type="boolean" default="True">
  If `False`, returns task\_id instead of blocking.
</ParamField>

<ParamField path="tags" type="dict[str, Any]" optional>
  Optional key-value pairs to tag and categorize the request. Tags are useful for organizing, filtering, and analyzing trace evaluations in Metabase or other 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:**

  ```python theme={null}
  tags={
      "environment": "production",
      "agent_version": "2.1.0",
      "experiment": "improved_prompts"
  }
  ```
</ParamField>

<ParamField path="evaluate_latest" type="boolean" optional>
  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.
</ParamField>

### Returns

<ResponseField name="result" type="MultiAgentTraceResponse | list[MultiAgentTraceResponse]">
  * Single or list of trace evaluation responses
  * Multiple criteria evaluated concurrently
</ResponseField>

### Example

```python theme={null}
async def evaluate_agent_trace():
    async with AsyncComposo() as client:
        # Assuming trace was captured using AgentTracer
        result = await client.evaluate_trace(
            trace=my_trace,
            criteria=[
                "Reward effective exploration",
                "Reward proper tool usage"
            ],
            tags={"environment": "production", "agent_version": "2.1.0"}
        )

        for res in result:
            print(f"Overall Score: {res.overall_score}")
            print(f"Agent Scores: {res.agent_scores}")

asyncio.run(evaluate_agent_trace())
```

***

## Context Manager Usage

The `AsyncComposo` client supports async context managers for automatic resource cleanup:

```python theme={null}
import asyncio
from composo import AsyncComposo

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

asyncio.run(main())
```

***

## Concurrency Control

The `AsyncComposo` client uses a semaphore to limit concurrent requests, preventing rate limit issues and excessive resource usage.

```python theme={null}
# Low concurrency (safer for rate limits)
client = AsyncComposo(max_concurrent_requests=5)

# Medium concurrency (balanced)
client = AsyncComposo(max_concurrent_requests=10)

# High concurrency (requires adequate rate limits)
client = AsyncComposo(max_concurrent_requests=20)
```

### Best Practices

1. **Start Conservative**: Begin with `max_concurrent_requests=5` and increase if needed
2. **Monitor Rate Limits**: Watch for `RateLimitError` exceptions and adjust accordingly
3. **Use Batching**: For very large datasets, process in batches to manage memory
4. **Handle Errors**: Use `asyncio.gather(..., return_exceptions=True)` for error resilience

***

## Performance Optimization

### Example: Optimal Batch Processing

```python theme={null}
from composo import AsyncComposo
import asyncio

async def optimized_evaluation(conversations, criteria):
    async with AsyncComposo(max_concurrent_requests=10) as client:
        # Use list comprehension for task creation
        tasks = [
            client.evaluate(messages=conv, criteria=criteria)
            for conv in conversations
        ]

        # Gather with error handling
        results = await asyncio.gather(*tasks, return_exceptions=True)

        # Process results and handle errors
        successes = []
        failures = []

        for i, result in enumerate(results):
            if isinstance(result, Exception):
                failures.append((i, result))
            else:
                successes.append(result)

        print(f"Success: {len(successes)}, Failures: {len(failures)}")
        return successes, failures

# Run
asyncio.run(optimized_evaluation(my_conversations, "Your criterion"))
```

***

## Comparison with Sync Client

| Feature             | `Composo`          | `AsyncComposo`            |
| ------------------- | ------------------ | ------------------------- |
| Use Case            | Single evaluations | Batch processing          |
| Concurrency         | Sequential         | Concurrent                |
| Performance         | Slower for batches | Optimized for batches     |
| API                 | Synchronous        | Asynchronous              |
| Complexity          | Simpler            | Requires async/await      |
| Concurrency Control | N/A                | `max_concurrent_requests` |

**When to use `AsyncComposo`:**

* Evaluating 10+ conversations
* Multiple criteria per evaluation
* High-throughput applications
* Integration with async frameworks (FastAPI, aiohttp)

**When to use `Composo`:**

* Single evaluations
* Simple scripts
* Synchronous applications
* Learning/prototyping