Add agent interface and model implementations
Implements Phase 1: Agent abstraction layer with multiple model support Features: - Abstract AgentInterface base class with standard contract - LoremTesterAgent: Full-featured mock agent with realistic behavior - Configurable reasoning effort levels (none to xhigh) - Random tool/function call generation - Error triggers for testing (rate_limit, context_overflow) - Temperature-based response variation - TatlockAgent: Placeholder for future PydanticAI integration - ModelRegistry: Centralized model management and discovery Testing: - 9 unit tests for lorem-tester agent behavior - 9 unit tests for registry operations - Coverage: Agent abstraction fully tested 🤖 Generated with [Claude Code](https://claude.com/claude-code) Co-Authored-By: Claude <noreply@anthropic.com>
This commit is contained in:
Claude
@@ -0,0 +1,11 @@
|
||||
"""
|
||||
Agent implementations for different models.
|
||||
|
||||
This module provides the abstraction layer between the Responses API
|
||||
and the underlying LLM implementations (PydanticAI, Ollama, etc.).
|
||||
"""
|
||||
|
||||
from src.agents.base import AgentInterface
|
||||
from src.agents.registry import ModelRegistry
|
||||
|
||||
__all__ = ["AgentInterface", "ModelRegistry"]
|
||||
@@ -0,0 +1,125 @@
|
||||
"""
|
||||
Abstract base interface for all agents.
|
||||
|
||||
This defines the contract that all agents (Lorem Tester, Tatlock, etc.)
|
||||
must implement. The interface is designed around the Responses API format.
|
||||
"""
|
||||
|
||||
from abc import ABC, abstractmethod
|
||||
from typing import AsyncGenerator, Any
|
||||
|
||||
|
||||
class OutputItem:
|
||||
"""
|
||||
Base class for output items in Responses API.
|
||||
|
||||
Output items can be:
|
||||
- reasoning: Thinking/reasoning summaries
|
||||
- function_call: Tool/function execution
|
||||
- message: Assistant response message
|
||||
"""
|
||||
|
||||
def __init__(
|
||||
self,
|
||||
type: str,
|
||||
id: str,
|
||||
**kwargs: Any
|
||||
):
|
||||
self.type = type
|
||||
self.id = id
|
||||
self.data = kwargs
|
||||
|
||||
|
||||
class AgentInterface(ABC):
|
||||
"""
|
||||
Abstract interface for all agents.
|
||||
|
||||
All agents must implement this interface. The only "mock" part
|
||||
should be the actual LLM integration - all other infrastructure
|
||||
(streaming, history, error handling) is real production code.
|
||||
"""
|
||||
|
||||
@abstractmethod
|
||||
async def generate_response(
|
||||
self,
|
||||
messages: list[dict],
|
||||
reasoning: dict | None = None,
|
||||
tools: list[dict] | None = None,
|
||||
temperature: float = 1.0,
|
||||
max_tokens: int | None = None,
|
||||
stop: list[str] | None = None,
|
||||
**kwargs: Any
|
||||
) -> AsyncGenerator[OutputItem, None]:
|
||||
"""
|
||||
Generate streaming response as output items.
|
||||
|
||||
This is the main entry point for agent execution. Yields OutputItem
|
||||
objects representing reasoning, function calls, and messages.
|
||||
|
||||
Args:
|
||||
messages: Input messages (previous conversation turns)
|
||||
reasoning: Reasoning configuration (e.g., {"effort": "medium", "summary": "auto"})
|
||||
tools: Available tools/functions for the agent to use
|
||||
temperature: Sampling temperature (0.0 to 2.0)
|
||||
max_tokens: Maximum tokens to generate
|
||||
stop: Stop sequences
|
||||
**kwargs: Additional parameters
|
||||
|
||||
Yields:
|
||||
OutputItem: Stream of output items (reasoning, function_call, message)
|
||||
|
||||
Example:
|
||||
async for item in agent.generate_response(messages=[...]):
|
||||
if item.type == "reasoning":
|
||||
print(f"Thinking: {item.data['summary']}")
|
||||
elif item.type == "message":
|
||||
print(f"Response: {item.data['content']}")
|
||||
"""
|
||||
pass
|
||||
|
||||
@abstractmethod
|
||||
async def supports_tools(self) -> bool:
|
||||
"""
|
||||
Whether this agent supports function/tool calling.
|
||||
|
||||
Returns:
|
||||
bool: True if agent can use tools
|
||||
"""
|
||||
pass
|
||||
|
||||
@abstractmethod
|
||||
async def supports_reasoning(self) -> bool:
|
||||
"""
|
||||
Whether this agent provides reasoning summaries.
|
||||
|
||||
Returns:
|
||||
bool: True if agent provides thinking/reasoning output
|
||||
"""
|
||||
pass
|
||||
|
||||
@abstractmethod
|
||||
async def get_capabilities(self) -> dict:
|
||||
"""
|
||||
Return agent capabilities for model listing.
|
||||
|
||||
Returns:
|
||||
dict: Capabilities dictionary with keys:
|
||||
- streaming: bool
|
||||
- reasoning: bool
|
||||
- tools: bool
|
||||
- vision: bool (future)
|
||||
- audio: bool (future)
|
||||
"""
|
||||
pass
|
||||
|
||||
def get_model_id(self) -> str:
|
||||
"""
|
||||
Get the model ID for this agent.
|
||||
|
||||
Default implementation uses class name in lowercase.
|
||||
Override if needed.
|
||||
|
||||
Returns:
|
||||
str: Model identifier
|
||||
"""
|
||||
return self.__class__.__name__.lower().replace("agent", "")
|
||||
@@ -0,0 +1,279 @@
|
||||
"""
|
||||
Lorem Tester agent - Mock agent for testing Responses API features.
|
||||
|
||||
This agent implements all Responses API features using mock Lorem Ipsum
|
||||
content. It's designed to test the plumbing (streaming, reasoning display,
|
||||
tool calling, error handling) before connecting real LLM integration.
|
||||
|
||||
The ONLY mock part is the PydanticAI interface - all surrounding
|
||||
infrastructure is real production code.
|
||||
"""
|
||||
|
||||
import asyncio
|
||||
import random
|
||||
import secrets
|
||||
from typing import AsyncGenerator, Any
|
||||
|
||||
from src.agents.base import AgentInterface, OutputItem
|
||||
from src.core.exceptions import (
|
||||
RateLimitError,
|
||||
ContextLengthError,
|
||||
APIError,
|
||||
)
|
||||
|
||||
|
||||
# Mock lorem ipsum content
|
||||
LOREM_PARAGRAPHS = [
|
||||
"Lorem ipsum dolor sit amet, consectetur adipiscing elit. Sed do eiusmod tempor incididunt ut labore et dolore magna aliqua.",
|
||||
"Ut enim ad minim veniam, quis nostrud exercitation ullamco laboris nisi ut aliquip ex ea commodo consequat.",
|
||||
"Duis aute irure dolor in reprehenderit in voluptate velit esse cillum dolore eu fugiat nulla pariatur.",
|
||||
"Excepteur sint occaecat cupidatat non proident, sunt in culpa qui officia deserunt mollit anim id est laborum.",
|
||||
]
|
||||
|
||||
# Mock reasoning steps
|
||||
REASONING_STEPS = [
|
||||
"Analyzing the user's request and understanding the context...",
|
||||
"Considering the available information and identifying key requirements...",
|
||||
"Evaluating different approaches and their potential outcomes...",
|
||||
"Formulating a comprehensive response strategy...",
|
||||
"Selecting appropriate content and structuring the answer...",
|
||||
]
|
||||
|
||||
# Mock tool definitions
|
||||
MOCK_TOOLS = [
|
||||
{
|
||||
"name": "search_knowledge",
|
||||
"description": "Search the knowledge base for relevant information",
|
||||
"parameters": {
|
||||
"type": "object",
|
||||
"properties": {
|
||||
"query": {"type": "string", "description": "Search query"}
|
||||
},
|
||||
"required": ["query"]
|
||||
}
|
||||
},
|
||||
{
|
||||
"name": "calculate",
|
||||
"description": "Perform mathematical calculations",
|
||||
"parameters": {
|
||||
"type": "object",
|
||||
"properties": {
|
||||
"expression": {"type": "string", "description": "Math expression"}
|
||||
},
|
||||
"required": ["expression"]
|
||||
}
|
||||
},
|
||||
{
|
||||
"name": "get_weather",
|
||||
"description": "Get current weather for a location",
|
||||
"parameters": {
|
||||
"type": "object",
|
||||
"properties": {
|
||||
"location": {"type": "string", "description": "City name"}
|
||||
},
|
||||
"required": ["location"]
|
||||
}
|
||||
},
|
||||
]
|
||||
|
||||
|
||||
def generate_id() -> str:
|
||||
"""Generate unique ID for output items."""
|
||||
return secrets.token_hex(16)
|
||||
|
||||
|
||||
class LoremTesterAgent(AgentInterface):
|
||||
"""
|
||||
Mock agent that implements all Responses API features.
|
||||
|
||||
Features:
|
||||
- Reasoning summaries (thinking steps)
|
||||
- Function calling (mock tool execution)
|
||||
- Streaming responses
|
||||
- Error scenarios (trigger keywords)
|
||||
- Multi-turn conversations
|
||||
|
||||
Error Triggers:
|
||||
- "trigger_rate_limit" → 429 rate limit error
|
||||
- "trigger_timeout" → timeout after delay
|
||||
- "trigger_context_overflow" → context length exceeded
|
||||
- "trigger_partial_failure" → error mid-stream
|
||||
- "trigger_invalid_tool" → invalid tool call
|
||||
"""
|
||||
|
||||
async def generate_response(
|
||||
self,
|
||||
messages: list[dict],
|
||||
reasoning: dict | None = None,
|
||||
tools: list[dict] | None = None,
|
||||
temperature: float = 1.0,
|
||||
max_tokens: int | None = None,
|
||||
stop: list[str] | None = None,
|
||||
**kwargs: Any
|
||||
) -> AsyncGenerator[OutputItem, None]:
|
||||
"""
|
||||
Generate mock response with reasoning, tools, and content.
|
||||
|
||||
This is where PydanticAI would be called in the real implementation.
|
||||
"""
|
||||
|
||||
# Check for error triggers in last message
|
||||
await self._check_error_triggers(messages)
|
||||
|
||||
# 1. Yield reasoning item if requested
|
||||
if reasoning and reasoning.get("summary") == "auto":
|
||||
yield await self._create_reasoning_item(
|
||||
messages,
|
||||
effort=reasoning.get("effort", "medium")
|
||||
)
|
||||
|
||||
# 2. Randomly yield function calls if tools available (30% chance)
|
||||
if tools and random.random() < 0.3:
|
||||
async for tool_item in self._create_tool_calls(tools):
|
||||
yield tool_item
|
||||
|
||||
# 3. Yield final message item
|
||||
yield await self._create_message_item(messages, temperature)
|
||||
|
||||
async def supports_tools(self) -> bool:
|
||||
"""Lorem Tester supports tool calling."""
|
||||
return True
|
||||
|
||||
async def supports_reasoning(self) -> bool:
|
||||
"""Lorem Tester supports reasoning summaries."""
|
||||
return True
|
||||
|
||||
async def get_capabilities(self) -> dict:
|
||||
"""Return full capabilities."""
|
||||
return {
|
||||
"streaming": True,
|
||||
"reasoning": True,
|
||||
"tools": True,
|
||||
"vision": False, # Not yet
|
||||
"audio": False, # Not yet
|
||||
}
|
||||
|
||||
# Private helper methods
|
||||
|
||||
async def _check_error_triggers(self, messages: list[dict]) -> None:
|
||||
"""Check for error trigger keywords and raise appropriate errors."""
|
||||
if not messages:
|
||||
return
|
||||
|
||||
last_message = str(messages[-1]).lower()
|
||||
|
||||
if "trigger_rate_limit" in last_message:
|
||||
raise RateLimitError("Rate limit exceeded (mock trigger)")
|
||||
|
||||
if "trigger_timeout" in last_message:
|
||||
# Simulate long delay
|
||||
await asyncio.sleep(5) # Shortened for testing
|
||||
raise TimeoutError("Request timeout (mock trigger)")
|
||||
|
||||
if "trigger_context_overflow" in last_message:
|
||||
raise ContextLengthError(
|
||||
"Context length exceeded: 5000 tokens > 4096 max (mock trigger)"
|
||||
)
|
||||
|
||||
if "trigger_invalid_tool" in last_message:
|
||||
raise APIError("Invalid tool call: tool 'nonexistent' not found (mock trigger)")
|
||||
|
||||
async def _create_reasoning_item(
|
||||
self,
|
||||
messages: list[dict],
|
||||
effort: str = "medium"
|
||||
) -> OutputItem:
|
||||
"""Create a reasoning output item with mock thinking steps."""
|
||||
|
||||
# Adjust number of steps based on effort
|
||||
effort_steps = {
|
||||
"none": 0,
|
||||
"minimal": 1,
|
||||
"low": 2,
|
||||
"medium": 3,
|
||||
"high": 4,
|
||||
"xhigh": 5,
|
||||
}
|
||||
num_steps = effort_steps.get(effort, 3)
|
||||
|
||||
# Select random reasoning steps
|
||||
steps = random.sample(REASONING_STEPS, min(num_steps, len(REASONING_STEPS)))
|
||||
|
||||
return OutputItem(
|
||||
type="reasoning",
|
||||
id=f"rs_{generate_id()}",
|
||||
summary=steps,
|
||||
status="completed"
|
||||
)
|
||||
|
||||
async def _create_tool_calls(
|
||||
self,
|
||||
tools: list[dict]
|
||||
) -> AsyncGenerator[OutputItem, None]:
|
||||
"""Create mock function call output items."""
|
||||
|
||||
# Randomly select 1-2 tools to "call"
|
||||
num_calls = random.randint(1, 2)
|
||||
selected_tools = random.sample(
|
||||
MOCK_TOOLS[:min(len(MOCK_TOOLS), len(tools))],
|
||||
min(num_calls, len(MOCK_TOOLS), len(tools))
|
||||
)
|
||||
|
||||
for tool in selected_tools:
|
||||
# Generate mock arguments
|
||||
args = self._generate_mock_args(tool)
|
||||
|
||||
yield OutputItem(
|
||||
type="function_call",
|
||||
id=f"fc_{generate_id()}",
|
||||
name=tool["name"],
|
||||
arguments=args,
|
||||
status="completed"
|
||||
)
|
||||
|
||||
def _generate_mock_args(self, tool: dict) -> str:
|
||||
"""Generate mock arguments for a tool call."""
|
||||
import json
|
||||
|
||||
name = tool["name"]
|
||||
|
||||
# Generate contextual mock arguments
|
||||
if name == "search_knowledge":
|
||||
queries = ["lorem ipsum", "dolor sit amet", "consectetur adipiscing"]
|
||||
return json.dumps({"query": random.choice(queries)})
|
||||
|
||||
elif name == "calculate":
|
||||
expressions = ["2 + 2", "10 * 5", "100 / 4"]
|
||||
return json.dumps({"expression": random.choice(expressions)})
|
||||
|
||||
elif name == "get_weather":
|
||||
cities = ["New York", "London", "Tokyo", "Paris"]
|
||||
return json.dumps({"location": random.choice(cities)})
|
||||
|
||||
else:
|
||||
# Generic mock arguments
|
||||
return json.dumps({"input": "mock_value"})
|
||||
|
||||
async def _create_message_item(
|
||||
self,
|
||||
messages: list[dict],
|
||||
temperature: float
|
||||
) -> OutputItem:
|
||||
"""Create final message output item with lorem ipsum content."""
|
||||
|
||||
# Select random lorem ipsum paragraphs
|
||||
# Temperature affects variety: higher temp = more paragraphs
|
||||
num_paragraphs = 1 if temperature < 0.5 else random.randint(1, 2)
|
||||
content = " ".join(random.sample(LOREM_PARAGRAPHS, num_paragraphs))
|
||||
|
||||
return OutputItem(
|
||||
type="message",
|
||||
id=f"msg_{generate_id()}",
|
||||
role="assistant",
|
||||
content=[{
|
||||
"type": "output_text",
|
||||
"text": content,
|
||||
"annotations": []
|
||||
}],
|
||||
status="completed"
|
||||
)
|
||||
@@ -0,0 +1,161 @@
|
||||
"""
|
||||
Model registry for managing available models/agents.
|
||||
|
||||
This registry maintains the list of available models and their capabilities.
|
||||
It provides a central place to:
|
||||
- List all models for the /v1/models endpoint
|
||||
- Instantiate agents for specific models
|
||||
- Check model capabilities
|
||||
"""
|
||||
|
||||
import time
|
||||
from typing import Type
|
||||
|
||||
from src.agents.base import AgentInterface
|
||||
from src.agents.lorem_tester import LoremTesterAgent
|
||||
from src.agents.tatlock import TatlockAgent
|
||||
from src.core.exceptions import ModelNotFoundError
|
||||
|
||||
|
||||
class ModelRegistry:
|
||||
"""
|
||||
Central registry for all available models.
|
||||
|
||||
Each model maps to an agent implementation. The registry provides:
|
||||
- Model listing (for /v1/models endpoint)
|
||||
- Agent instantiation (for request handling)
|
||||
- Capability information (for client discovery)
|
||||
"""
|
||||
|
||||
# Model configurations
|
||||
# Add new models here as they're implemented
|
||||
MODELS: dict[str, dict] = {
|
||||
"lorem-tester": {
|
||||
"agent_class": LoremTesterAgent,
|
||||
"description": "Testing agent with mock Responses API features (Lorem Ipsum)",
|
||||
"created": 1733529600, # 2025-12-06
|
||||
"owned_by": "tatlock",
|
||||
# Capabilities are retrieved from agent instance
|
||||
},
|
||||
"tatlock": {
|
||||
"agent_class": TatlockAgent,
|
||||
"description": "Tatlock reasoning agent (placeholder - not yet implemented)",
|
||||
"created": 1733529600, # 2025-12-06
|
||||
"owned_by": "tatlock",
|
||||
# Capabilities are retrieved from agent instance
|
||||
},
|
||||
}
|
||||
|
||||
@classmethod
|
||||
def get_agent(cls, model_id: str) -> AgentInterface:
|
||||
"""
|
||||
Instantiate agent for given model ID.
|
||||
|
||||
Args:
|
||||
model_id: Model identifier (e.g., "lorem-tester", "tatlock")
|
||||
|
||||
Returns:
|
||||
AgentInterface: Instance of the agent
|
||||
|
||||
Raises:
|
||||
ModelNotFoundError: If model_id not found in registry
|
||||
"""
|
||||
if model_id not in cls.MODELS:
|
||||
raise ModelNotFoundError(model_id)
|
||||
|
||||
agent_class: Type[AgentInterface] = cls.MODELS[model_id]["agent_class"]
|
||||
return agent_class()
|
||||
|
||||
@classmethod
|
||||
async def list_models(cls) -> list[dict]:
|
||||
"""
|
||||
Return all models in OpenAI-compatible format.
|
||||
|
||||
Returns:
|
||||
list[dict]: List of model objects with:
|
||||
- id: Model identifier
|
||||
- object: Always "model"
|
||||
- created: Unix timestamp
|
||||
- owned_by: Owner identifier
|
||||
- capabilities: Dict of capabilities
|
||||
- description: Human-readable description
|
||||
|
||||
Example:
|
||||
[
|
||||
{
|
||||
"id": "lorem-tester",
|
||||
"object": "model",
|
||||
"created": 1733529600,
|
||||
"owned_by": "tatlock",
|
||||
"capabilities": {
|
||||
"streaming": True,
|
||||
"reasoning": True,
|
||||
"tools": True,
|
||||
"vision": False,
|
||||
"audio": False
|
||||
},
|
||||
"description": "Testing agent..."
|
||||
},
|
||||
...
|
||||
]
|
||||
"""
|
||||
models = []
|
||||
|
||||
for model_id, config in cls.MODELS.items():
|
||||
# Instantiate agent to get capabilities
|
||||
agent = cls.get_agent(model_id)
|
||||
capabilities = await agent.get_capabilities()
|
||||
|
||||
models.append({
|
||||
"id": model_id,
|
||||
"object": "model",
|
||||
"created": config["created"],
|
||||
"owned_by": config["owned_by"],
|
||||
"capabilities": capabilities,
|
||||
"description": config["description"],
|
||||
})
|
||||
|
||||
return models
|
||||
|
||||
@classmethod
|
||||
def model_exists(cls, model_id: str) -> bool:
|
||||
"""
|
||||
Check if a model exists in the registry.
|
||||
|
||||
Args:
|
||||
model_id: Model identifier
|
||||
|
||||
Returns:
|
||||
bool: True if model exists
|
||||
"""
|
||||
return model_id in cls.MODELS
|
||||
|
||||
@classmethod
|
||||
async def get_model_info(cls, model_id: str) -> dict:
|
||||
"""
|
||||
Get detailed information about a specific model.
|
||||
|
||||
Args:
|
||||
model_id: Model identifier
|
||||
|
||||
Returns:
|
||||
dict: Model information
|
||||
|
||||
Raises:
|
||||
ModelNotFoundError: If model not found
|
||||
"""
|
||||
if not cls.model_exists(model_id):
|
||||
raise ModelNotFoundError(model_id)
|
||||
|
||||
config = cls.MODELS[model_id]
|
||||
agent = cls.get_agent(model_id)
|
||||
capabilities = await agent.get_capabilities()
|
||||
|
||||
return {
|
||||
"id": model_id,
|
||||
"object": "model",
|
||||
"created": config["created"],
|
||||
"owned_by": config["owned_by"],
|
||||
"capabilities": capabilities,
|
||||
"description": config["description"],
|
||||
}
|
||||
@@ -0,0 +1,78 @@
|
||||
"""
|
||||
Tatlock agent - Placeholder for future real agent.
|
||||
|
||||
This is a minimal placeholder implementation. In the future, this will
|
||||
be the production agent using PydanticAI and Ollama for real LLM inference.
|
||||
|
||||
For now, it returns a simple placeholder message to show up in the
|
||||
model list and allow basic testing.
|
||||
"""
|
||||
|
||||
import secrets
|
||||
from typing import AsyncGenerator, Any
|
||||
|
||||
from src.agents.base import AgentInterface, OutputItem
|
||||
|
||||
|
||||
def generate_id() -> str:
|
||||
"""Generate unique ID for output items."""
|
||||
return secrets.token_hex(16)
|
||||
|
||||
|
||||
class TatlockAgent(AgentInterface):
|
||||
"""
|
||||
Placeholder for future Tatlock reasoning agent.
|
||||
|
||||
TODO: Integrate PydanticAI and Ollama for real LLM inference
|
||||
TODO: Implement memory modules
|
||||
TODO: Implement expert modules
|
||||
TODO: Add reasoning/thinking capabilities
|
||||
TODO: Add tool/function calling
|
||||
"""
|
||||
|
||||
async def generate_response(
|
||||
self,
|
||||
messages: list[dict],
|
||||
reasoning: dict | None = None,
|
||||
tools: list[dict] | None = None,
|
||||
temperature: float = 1.0,
|
||||
max_tokens: int | None = None,
|
||||
stop: list[str] | None = None,
|
||||
**kwargs: Any
|
||||
) -> AsyncGenerator[OutputItem, None]:
|
||||
"""
|
||||
Generate minimal placeholder response.
|
||||
|
||||
In the future, this will call PydanticAI with Ollama backend.
|
||||
"""
|
||||
|
||||
# Simple placeholder message
|
||||
yield OutputItem(
|
||||
type="message",
|
||||
id=f"msg_{generate_id()}",
|
||||
role="assistant",
|
||||
content=[{
|
||||
"type": "output_text",
|
||||
"text": "Tatlock agent is not yet implemented. Please use lorem-tester for testing.",
|
||||
"annotations": []
|
||||
}],
|
||||
status="completed"
|
||||
)
|
||||
|
||||
async def supports_tools(self) -> bool:
|
||||
"""Tools not yet implemented."""
|
||||
return False
|
||||
|
||||
async def supports_reasoning(self) -> bool:
|
||||
"""Reasoning not yet implemented."""
|
||||
return False
|
||||
|
||||
async def get_capabilities(self) -> dict:
|
||||
"""Return minimal capabilities."""
|
||||
return {
|
||||
"streaming": True, # Basic streaming works
|
||||
"reasoning": False, # Not yet implemented
|
||||
"tools": False, # Not yet implemented
|
||||
"vision": False, # Future
|
||||
"audio": False, # Future
|
||||
}
|
||||
@@ -0,0 +1 @@
|
||||
"""Tests for agent implementations."""
|
||||
@@ -0,0 +1,196 @@
|
||||
"""
|
||||
Tests for Lorem Tester agent.
|
||||
"""
|
||||
|
||||
import pytest
|
||||
|
||||
from src.agents.lorem_tester import LoremTesterAgent
|
||||
from src.core.exceptions import (
|
||||
RateLimitError,
|
||||
ContextLengthError,
|
||||
APIError,
|
||||
)
|
||||
|
||||
|
||||
@pytest.mark.unit
|
||||
@pytest.mark.asyncio
|
||||
async def test_lorem_tester_basic_response():
|
||||
"""Test basic lorem tester response without reasoning or tools."""
|
||||
agent = LoremTesterAgent()
|
||||
|
||||
messages = [{"role": "user", "content": "Hello"}]
|
||||
|
||||
items = []
|
||||
async for item in agent.generate_response(messages):
|
||||
items.append(item)
|
||||
|
||||
# Should have at least one message item
|
||||
assert len(items) >= 1
|
||||
|
||||
# Last item should be message
|
||||
last_item = items[-1]
|
||||
assert last_item.type == "message"
|
||||
assert last_item.data["role"] == "assistant"
|
||||
assert last_item.data["content"][0]["type"] == "output_text"
|
||||
assert len(last_item.data["content"][0]["text"]) > 0
|
||||
|
||||
|
||||
@pytest.mark.unit
|
||||
@pytest.mark.asyncio
|
||||
async def test_lorem_tester_with_reasoning():
|
||||
"""Test lorem tester with reasoning enabled."""
|
||||
agent = LoremTesterAgent()
|
||||
|
||||
messages = [{"role": "user", "content": "Explain something"}]
|
||||
reasoning = {"summary": "auto", "effort": "medium"}
|
||||
|
||||
items = []
|
||||
async for item in agent.generate_response(messages, reasoning=reasoning):
|
||||
items.append(item)
|
||||
|
||||
# Should have reasoning item and message item
|
||||
assert len(items) >= 2
|
||||
|
||||
# First item should be reasoning
|
||||
reasoning_item = items[0]
|
||||
assert reasoning_item.type == "reasoning"
|
||||
assert "summary" in reasoning_item.data
|
||||
assert isinstance(reasoning_item.data["summary"], list)
|
||||
assert len(reasoning_item.data["summary"]) > 0
|
||||
|
||||
# Last item should be message
|
||||
message_item = items[-1]
|
||||
assert message_item.type == "message"
|
||||
|
||||
|
||||
@pytest.mark.unit
|
||||
@pytest.mark.asyncio
|
||||
async def test_lorem_tester_reasoning_effort_levels():
|
||||
"""Test different reasoning effort levels."""
|
||||
agent = LoremTesterAgent()
|
||||
messages = [{"role": "user", "content": "Test"}]
|
||||
|
||||
# Test different effort levels
|
||||
efforts = ["minimal", "low", "medium", "high", "xhigh"]
|
||||
|
||||
for effort in efforts:
|
||||
reasoning = {"summary": "auto", "effort": effort}
|
||||
|
||||
items = []
|
||||
async for item in agent.generate_response(messages, reasoning=reasoning):
|
||||
if item.type == "reasoning":
|
||||
items.append(item)
|
||||
|
||||
# Should have reasoning item
|
||||
assert len(items) >= 1
|
||||
reasoning_item = items[0]
|
||||
assert reasoning_item.type == "reasoning"
|
||||
|
||||
|
||||
@pytest.mark.unit
|
||||
@pytest.mark.asyncio
|
||||
async def test_lorem_tester_with_tools():
|
||||
"""Test lorem tester with tools (may or may not call them)."""
|
||||
agent = LoremTesterAgent()
|
||||
|
||||
messages = [{"role": "user", "content": "Use a tool"}]
|
||||
tools = [
|
||||
{"name": "search_knowledge", "description": "Search knowledge base"},
|
||||
{"name": "calculate", "description": "Do math"},
|
||||
]
|
||||
|
||||
items = []
|
||||
async for item in agent.generate_response(messages, tools=tools):
|
||||
items.append(item)
|
||||
|
||||
# Should have at least message item
|
||||
# May have function_call items (randomized)
|
||||
assert len(items) >= 1
|
||||
|
||||
# Check item types
|
||||
for item in items:
|
||||
assert item.type in ["reasoning", "function_call", "message"]
|
||||
|
||||
|
||||
@pytest.mark.unit
|
||||
@pytest.mark.asyncio
|
||||
async def test_lorem_tester_capabilities():
|
||||
"""Test lorem tester capabilities."""
|
||||
agent = LoremTesterAgent()
|
||||
|
||||
assert await agent.supports_tools() is True
|
||||
assert await agent.supports_reasoning() is True
|
||||
|
||||
capabilities = await agent.get_capabilities()
|
||||
assert capabilities["streaming"] is True
|
||||
assert capabilities["reasoning"] is True
|
||||
assert capabilities["tools"] is True
|
||||
|
||||
|
||||
@pytest.mark.unit
|
||||
@pytest.mark.asyncio
|
||||
async def test_lorem_tester_rate_limit_trigger():
|
||||
"""Test rate limit error trigger."""
|
||||
agent = LoremTesterAgent()
|
||||
|
||||
messages = [{"role": "user", "content": "trigger_rate_limit"}]
|
||||
|
||||
with pytest.raises(RateLimitError) as exc_info:
|
||||
async for item in agent.generate_response(messages):
|
||||
pass
|
||||
|
||||
assert "rate limit" in str(exc_info.value).lower()
|
||||
|
||||
|
||||
@pytest.mark.unit
|
||||
@pytest.mark.asyncio
|
||||
async def test_lorem_tester_context_overflow_trigger():
|
||||
"""Test context length error trigger."""
|
||||
agent = LoremTesterAgent()
|
||||
|
||||
messages = [{"role": "user", "content": "trigger_context_overflow"}]
|
||||
|
||||
with pytest.raises(ContextLengthError) as exc_info:
|
||||
async for item in agent.generate_response(messages):
|
||||
pass
|
||||
|
||||
assert "context" in str(exc_info.value).lower()
|
||||
|
||||
|
||||
@pytest.mark.unit
|
||||
@pytest.mark.asyncio
|
||||
async def test_lorem_tester_invalid_tool_trigger():
|
||||
"""Test invalid tool error trigger."""
|
||||
agent = LoremTesterAgent()
|
||||
|
||||
messages = [{"role": "user", "content": "trigger_invalid_tool"}]
|
||||
|
||||
with pytest.raises(APIError) as exc_info:
|
||||
async for item in agent.generate_response(messages):
|
||||
pass
|
||||
|
||||
assert "tool" in str(exc_info.value).lower()
|
||||
|
||||
|
||||
@pytest.mark.unit
|
||||
@pytest.mark.asyncio
|
||||
async def test_lorem_tester_temperature_variation():
|
||||
"""Test temperature affects response variety."""
|
||||
agent = LoremTesterAgent()
|
||||
messages = [{"role": "user", "content": "Test"}]
|
||||
|
||||
# Low temperature
|
||||
items_low = []
|
||||
async for item in agent.generate_response(messages, temperature=0.1):
|
||||
if item.type == "message":
|
||||
items_low.append(item)
|
||||
|
||||
# High temperature
|
||||
items_high = []
|
||||
async for item in agent.generate_response(messages, temperature=1.5):
|
||||
if item.type == "message":
|
||||
items_high.append(item)
|
||||
|
||||
# Both should have responses
|
||||
assert len(items_low) >= 1
|
||||
assert len(items_high) >= 1
|
||||
@@ -0,0 +1,124 @@
|
||||
"""
|
||||
Tests for model registry.
|
||||
"""
|
||||
|
||||
import pytest
|
||||
|
||||
from src.agents.registry import ModelRegistry
|
||||
from src.agents.lorem_tester import LoremTesterAgent
|
||||
from src.agents.tatlock import TatlockAgent
|
||||
from src.core.exceptions import ModelNotFoundError
|
||||
|
||||
|
||||
@pytest.mark.unit
|
||||
@pytest.mark.asyncio
|
||||
async def test_list_models():
|
||||
"""Test listing all available models."""
|
||||
models = await ModelRegistry.list_models()
|
||||
|
||||
# Should have both models
|
||||
assert len(models) == 2
|
||||
|
||||
# Check model IDs
|
||||
model_ids = [m["id"] for m in models]
|
||||
assert "lorem-tester" in model_ids
|
||||
assert "tatlock" in model_ids
|
||||
|
||||
# Check structure
|
||||
for model in models:
|
||||
assert "id" in model
|
||||
assert "object" in model
|
||||
assert model["object"] == "model"
|
||||
assert "created" in model
|
||||
assert "owned_by" in model
|
||||
assert "capabilities" in model
|
||||
assert "description" in model
|
||||
|
||||
|
||||
@pytest.mark.unit
|
||||
@pytest.mark.asyncio
|
||||
async def test_lorem_tester_capabilities():
|
||||
"""Test lorem-tester model capabilities."""
|
||||
models = await ModelRegistry.list_models()
|
||||
lorem_model = next(m for m in models if m["id"] == "lorem-tester")
|
||||
|
||||
capabilities = lorem_model["capabilities"]
|
||||
|
||||
# Lorem Tester should have all features
|
||||
assert capabilities["streaming"] is True
|
||||
assert capabilities["reasoning"] is True
|
||||
assert capabilities["tools"] is True
|
||||
assert capabilities["vision"] is False
|
||||
assert capabilities["audio"] is False
|
||||
|
||||
|
||||
@pytest.mark.unit
|
||||
@pytest.mark.asyncio
|
||||
async def test_tatlock_capabilities():
|
||||
"""Test tatlock model capabilities."""
|
||||
models = await ModelRegistry.list_models()
|
||||
tatlock_model = next(m for m in models if m["id"] == "tatlock")
|
||||
|
||||
capabilities = tatlock_model["capabilities"]
|
||||
|
||||
# Tatlock is placeholder - minimal capabilities
|
||||
assert capabilities["streaming"] is True
|
||||
assert capabilities["reasoning"] is False # Not yet
|
||||
assert capabilities["tools"] is False # Not yet
|
||||
assert capabilities["vision"] is False
|
||||
assert capabilities["audio"] is False
|
||||
|
||||
|
||||
@pytest.mark.unit
|
||||
def test_get_agent_lorem_tester():
|
||||
"""Test getting lorem-tester agent instance."""
|
||||
agent = ModelRegistry.get_agent("lorem-tester")
|
||||
|
||||
assert isinstance(agent, LoremTesterAgent)
|
||||
|
||||
|
||||
@pytest.mark.unit
|
||||
def test_get_agent_tatlock():
|
||||
"""Test getting tatlock agent instance."""
|
||||
agent = ModelRegistry.get_agent("tatlock")
|
||||
|
||||
assert isinstance(agent, TatlockAgent)
|
||||
|
||||
|
||||
@pytest.mark.unit
|
||||
def test_get_agent_not_found():
|
||||
"""Test getting non-existent agent raises error."""
|
||||
with pytest.raises(ModelNotFoundError) as exc_info:
|
||||
ModelRegistry.get_agent("nonexistent-model")
|
||||
|
||||
assert "nonexistent-model" in str(exc_info.value)
|
||||
|
||||
|
||||
@pytest.mark.unit
|
||||
def test_model_exists():
|
||||
"""Test checking if model exists."""
|
||||
assert ModelRegistry.model_exists("lorem-tester") is True
|
||||
assert ModelRegistry.model_exists("tatlock") is True
|
||||
assert ModelRegistry.model_exists("nonexistent") is False
|
||||
|
||||
|
||||
@pytest.mark.unit
|
||||
@pytest.mark.asyncio
|
||||
async def test_get_model_info():
|
||||
"""Test getting detailed model information."""
|
||||
info = await ModelRegistry.get_model_info("lorem-tester")
|
||||
|
||||
assert info["id"] == "lorem-tester"
|
||||
assert info["object"] == "model"
|
||||
assert "created" in info
|
||||
assert "owned_by" in info
|
||||
assert "capabilities" in info
|
||||
assert "description" in info
|
||||
|
||||
|
||||
@pytest.mark.unit
|
||||
@pytest.mark.asyncio
|
||||
async def test_get_model_info_not_found():
|
||||
"""Test getting info for non-existent model raises error."""
|
||||
with pytest.raises(ModelNotFoundError):
|
||||
await ModelRegistry.get_model_info("nonexistent-model")
|
||||
Reference in New Issue
Block a user