Models

AgentOpera provides a flexible and powerful abstraction for working with large language models (LLMs) through its models module. This module implements a standard protocol for interacting with various LLM providers, making it easy to switch between different models or providers while maintaining a consistent interface.

ChatCompletionClient Protocol

At the core of the models module is the ChatCompletionClient abstract base class, which defines the protocol that all model clients must implement. This protocol standardizes how applications interact with language models:

class ChatCompletionClient(ABC):
    @abstractmethod
    async def create(
        self,
        messages: Sequence[LLMMessage],
        *,
        tools: Sequence[Tool | ToolSchema] = [],
        json_output: Optional[bool] = None,
        extra_create_args: Mapping[str, Any] = {},
        cancellation_token: Optional[CancellationToken] = None,
    ) -> CreateResult: ...

    @abstractmethod
    def create_stream(
        self,
        messages: Sequence[LLMMessage],
        *,
        tools: Sequence[Tool | ToolSchema] = [],
        json_output: Optional[bool] = None,
        extra_create_args: Mapping[str, Any] = {},
        cancellation_token: Optional[CancellationToken] = None,
    ) -> AsyncGenerator[Union[str, CreateResult], None]: ...

    @property
    @abstractmethod
    def model_info(self) -> ModelInfo: ...

Key Methods

• create() - Makes a synchronous call to the LLM and returns the complete response.

• create_stream() - Makes a streaming call to the LLM, yielding results as they become available.

• model_info - Returns information about the model's capabilities.

Model Information and Capabilities

AgentOpera uses the ModelInfo type to represent model capabilities:

class ModelInfo(TypedDict, total=False):
    vision: Required[bool]           # Support for image inputs
    function_calling: Required[bool] # Support for function calling
    json_output: Required[bool]      # Support for JSON output
    family: Required[ModelFamily.ANY | str] # Model family information

The ModelFamily class provides constants for common model families:

class ModelFamily:
    GPT_4O = "gpt-4o"
    GPT_4 = "gpt-4"
    R1 = "r1"
    GEMINI_1_5_PRO = "gemini-1.5-pro"
    CLAUDE_3_7_SONNET = "claude-3.7-sonnet"
    UNKNOWN = "unknown"

OpenAIChatCompletionClient

The OpenAIChatCompletionClient is the commonly used implementation of the ChatCompletionClient protocol, providing access to OpenAI's models.

Basic Usage

from agentopera.models.openai import OpenAIChatCompletionClient
from agentopera.engine.types.models import UserMessage

# Initialize the client
openai_client = OpenAIChatCompletionClient(
    model="gpt-4o-2024-08-06",
    # api_key="sk-..." # Optional if OPENAI_API_KEY environment variable is set
)

# Make a request
async def get_response():
    result = await openai_client.create([
        UserMessage(content="What is the capital of France?", source="user")
    ])
    print(result.content)

# For streaming responses
async def stream_response():
    async for chunk in openai_client.create_stream([
        UserMessage(content="Tell me about Paris", source="user")
    ]):
        if isinstance(chunk, str):
            print(chunk, end="")
        else:
            # Final result with usage info
            print(f"\nUsage: {chunk.usage}")

Function Calling

The client supports function calling through the tools parameter:

from agentopera.engine.function_call import Tool

# Define a function tool
weather_tool = Tool(
    name="get_weather",
    description="Get the current weather for a location",
    parameters={
        "type": "object",
        "properties": {
            "location": {
                "type": "string",
                "description": "The city and state, e.g. San Francisco, CA"
            }
        },
        "required": ["location"]
    }
)

# Use it in a request
async def use_function():
    result = await openai_client.create(
        [UserMessage(content="What's the weather in Paris?", source="user")],
        tools=[weather_tool]
    )
    
    # Check if the model called a function
    if isinstance(result.content, list):
        func_call = result.content[0]
        print(f"Function: {func_call.name}")
        print(f"Arguments: {func_call.arguments}")

Other Model Clients

AgentOpera also provides clients for other LLM providers:

AzureOpenAIChatCompletionClient

For Azure OpenAI-hosted models:

from agentopera.models.openai import AzureOpenAIChatCompletionClient
from azure.identity import DefaultAzureCredential, get_bearer_token_provider

# Using AAD authentication
token_provider = get_bearer_token_provider(
    DefaultAzureCredential(), 
    "https://cognitiveservices.azure.com/.default"
)

azure_client = AzureOpenAIChatCompletionClient(
    azure_deployment="your-deployment-name",
    model="gpt-4o",
    api_version="2024-06-01",
    azure_endpoint="https://your-endpoint.openai.azure.com/",
    azure_ad_token_provider=token_provider
)

AnthropicChatCompletionClient

For Claude and other Anthropic models:

from agentopera.models.anthropic import AnthropicChatCompletionClient

anthropic_client = AnthropicChatCompletionClient(
    model="claude-3-sonnet-20240229",
    # api_key="sk-..." # Optional if ANTHROPIC_API_KEY environment variable is set
)

OllamaChatCompletionClient

For locally hosted models using Ollama:

from agentopera.models.ollama import OllamaChatCompletionClient

ollama_client = OllamaChatCompletionClient(
    model="deepseek-r1:1.5b",
    base_url="http://localhost:11434"
)

Advanced Features

Cancellation

Long-running requests can be cancelled using a CancellationToken:

from agentopera.engine.types.agent import CancellationToken

async def cancellable_request():
    token = CancellationToken()
    
    # Start the request
    task = asyncio.create_task(
        openai_client.create(
            [UserMessage(content="Write a 10-page essay on...", source="user")],
            cancellation_token=token
        )
    )
    
    # Cancel after 5 seconds
    await asyncio.sleep(5)
    token.cancel()
    
    try:
        result = await task
    except asyncio.CancelledError:
        print("Request was cancelled")