tessl/pypi-llm

CLI utility and Python library for interacting with Large Language Models from multiple providers including OpenAI, Anthropic, Google, and Meta plus locally installed models.

—

Pending

Overview

Eval results

Files

Models and Conversations

Name: tessl/pypi-llm
Author: tessl

Core model management, conversation handling, prompt processing, and response streaming functionality. This module provides both synchronous and asynchronous APIs for interacting with Large Language Models from multiple providers.

Capabilities

Model Discovery and Management

The package provides functions to discover, retrieve, and manage models from various providers through a plugin-based registration system.

def get_model(name: Optional[str] = None) -> Model:
    """
    Get a model by name or alias. Returns the default model if name is None.
    
    Args:
        name: Model name or alias. Uses default if None.
        
    Returns:
        Model instance
        
    Raises:
        UnknownModelError: If model name/alias not found
    """

def get_async_model(name: Optional[str] = None) -> AsyncModel:
    """
    Get an async model by name or alias. Returns default async model if name is None.
    
    Args:
        name: Model name or alias. Uses default if None.
        
    Returns:
        AsyncModel instance
        
    Raises:
        UnknownModelError: If async model name/alias not found
    """

def get_models() -> List[Model]:
    """Get all registered synchronous models."""

def get_async_models() -> List[AsyncModel]:
    """Get all registered asynchronous models."""

def get_models_with_aliases() -> List[ModelWithAliases]:
    """Get models with their configured aliases."""

def get_model_aliases() -> Dict[str, Model]:
    """Get mapping of all aliases to their corresponding models."""

def get_async_model_aliases() -> Dict[str, AsyncModel]:
    """Get mapping of all aliases to their corresponding async models."""

Model Hierarchy

The package defines an abstract model hierarchy supporting both synchronous and asynchronous operations.

class Model(ABC):
    """Abstract base class for synchronous LLM models."""
    model_id: str
    
    @abstractmethod
    def prompt(self, prompt: Prompt, **kwargs) -> Response:
        """Execute a prompt and return response."""
    
    def conversation(self, **kwargs) -> Conversation:
        """Create a new conversation with this model."""

class KeyModel(Model):
    """Abstract base for models requiring API keys."""
    needs_key: str
    key_env_var: Optional[str] = None
    
    def __init__(self, key: Optional[str] = None):
        """Initialize with optional API key."""

class AsyncModel(ABC):
    """Abstract base class for asynchronous LLM models."""
    model_id: str
    
    @abstractmethod
    async def prompt(self, prompt: Prompt, **kwargs) -> AsyncResponse:
        """Execute a prompt asynchronously and return response."""
    
    def conversation(self, **kwargs) -> AsyncConversation:
        """Create a new async conversation with this model."""

class AsyncKeyModel(AsyncModel):
    """Abstract base for async models requiring API keys."""
    needs_key: str
    key_env_var: Optional[str] = None
    
    def __init__(self, key: Optional[str] = None):
        """Initialize with optional API key."""

class ModelWithAliases:
    """Container for a model with its aliases."""
    model: Model
    async_model: Optional[AsyncModel]
    aliases: List[str]

class Options(BaseModel):
    """Pydantic model for model configuration options."""
    model_config = ConfigDict(extra="forbid")

Conversation Management

Conversations maintain state and history across multiple prompt-response interactions.

class Conversation:
    """Synchronous conversation management with tool chaining support."""
    
    def __init__(self, model: Optional[Model] = None, conversation_id: Optional[str] = None):
        """
        Initialize conversation.
        
        Args:
            model: Model to use for conversation
            conversation_id: Optional conversation ID for persistence
        """
    
    def prompt(
        self, 
        prompt: Union[str, Prompt], 
        *, 
        attachments: Optional[List[Attachment]] = None,
        tools: Optional[List[Tool]] = None,
        stream: bool = False,
        **kwargs
    ) -> Response:
        """
        Send a prompt and get response.
        
        Args:
            prompt: Text prompt or Prompt object
            attachments: Optional file/URL attachments
            tools: Optional tools for function calling
            stream: Whether to stream response
            **kwargs: Additional model options
            
        Returns:
            Response object with text and metadata
        """
    
    def responses(self) -> List[Response]:
        """Get all responses in conversation history."""

class AsyncConversation:
    """Asynchronous conversation management with tool chaining support."""
    
    def __init__(self, model: Optional[AsyncModel] = None, conversation_id: Optional[str] = None):
        """Initialize async conversation."""
    
    async def prompt(
        self,
        prompt: Union[str, Prompt],
        *,
        attachments: Optional[List[Attachment]] = None,
        tools: Optional[List[Tool]] = None,
        stream: bool = False,
        **kwargs
    ) -> AsyncResponse:
        """Send a prompt asynchronously and get response."""
    
    async def responses(self) -> List[AsyncResponse]:
        """Get all responses in conversation history."""

Prompt System

The Prompt class provides a rich container for input data including text, attachments, tools, and structured schemas.

class Prompt:
    """Container for prompts with tools, schema, and attachments."""
    
    def __init__(
        self,
        prompt: Optional[str] = None,
        *,
        system: Optional[str] = None,
        attachments: Optional[List[Attachment]] = None,
        tools: Optional[List[Tool]] = None,
        tool_choice: Optional[str] = None,
        schema: Optional[dict] = None,
        options: Optional[Options] = None
    ):
        """
        Initialize prompt.
        
        Args:
            prompt: Main prompt text
            system: System message
            attachments: File/URL attachments
            tools: Available tools for function calling
            tool_choice: Tool selection strategy
            schema: JSON schema for structured output
            options: Model configuration options
        """
    
    def __str__(self) -> str:
        """Return prompt text."""

class Attachment:
    """File/URL/binary content attachment to prompts."""
    
    def __init__(
        self,
        type: Optional[str] = None,
        path: Optional[str] = None,
        url: Optional[str] = None,
        content: Optional[bytes] = None
    ):
        """
        Initialize attachment.
        
        Args:
            type: MIME type
            path: File path
            url: URL to content
            content: Binary content
        """
    
    def id(self) -> str:
        """Generate content hash ID."""
    
    def resolve_type(self) -> str:
        """Determine MIME type from content/path/URL."""
    
    def content_bytes(self) -> bytes:
        """Get content as bytes."""
    
    def base64_content(self) -> str:
        """Get base64-encoded content."""

Response Handling

Response objects provide access to model output with support for streaming, tool calls, and metadata.

class Response:
    """Synchronous response object with streaming support."""
    
    def text(self) -> str:
        """Get complete response text."""
    
    def __iter__(self) -> Iterator[str]:
        """Stream response text chunks."""
    
    def tool_calls(self) -> List[ToolCall]:
        """Get tool calls made during response."""
    
    def usage(self) -> Usage:
        """Get token usage information."""
    
    def response_json(self) -> Optional[dict]:
        """Get structured JSON response if schema was provided."""

class AsyncResponse:
    """Asynchronous response object for streaming/awaitable results."""
    
    async def text(self) -> str:
        """Get complete response text asynchronously."""
    
    def __aiter__(self) -> AsyncIterator[str]:
        """Stream response text chunks asynchronously."""
    
    async def tool_calls(self) -> List[ToolCall]:
        """Get tool calls made during response."""
    
    async def usage(self) -> Usage:
        """Get token usage information."""
    
    async def response_json(self) -> Optional[dict]:
        """Get structured JSON response if schema was provided."""

class ChainResponse(Response):
    """Synchronous chained tool execution response."""
    
    def responses(self) -> List[Response]:
        """Get all responses in the tool chain."""

class AsyncChainResponse(AsyncResponse):
    """Asynchronous chained tool execution response."""
    
    async def responses(self) -> List[AsyncResponse]:
        """Get all responses in the tool chain."""

Usage Tracking

The Usage class tracks token consumption and provides detailed usage statistics.

class Usage:
    """Token usage tracking with detailed breakdown."""
    
    def __init__(
        self,
        input: Optional[int] = None,
        output: Optional[int] = None,
        details: Optional[Dict[str, Any]] = None
    ):
        """
        Initialize usage tracking.
        
        Args:
            input: Input token count
            output: Output token count
            details: Additional usage details
        """
    
    input: Optional[int]
    output: Optional[int]
    details: Optional[Dict[str, Any]]

Error Handling

The models module defines several exception classes for error handling:

class ModelError(Exception):
    """Base exception for model-related errors."""

class NeedsKeyException(ModelError):
    """Raised when API key is required but missing."""

class UnknownModelError(KeyError):
    """Raised when model name or alias is not found."""

Usage Examples

Basic Model Usage

import llm

# Get default model and send prompt
model = llm.get_model()
response = model.prompt("What is machine learning?")
print(response.text())

# Use specific model
model = llm.get_model("gpt-4")
response = model.prompt("Explain quantum computing in simple terms")
print(response.text())

Streaming Responses

import llm

model = llm.get_model()
response = model.prompt("Write a short story", stream=True)

# Stream response chunks
for chunk in response:
    print(chunk, end="", flush=True)
print()  # Final newline

Conversation with History

import llm

model = llm.get_model("claude-3-sonnet")
conversation = model.conversation()

# Multi-turn conversation
response1 = conversation.prompt("My name is Alice. What's a good hobby?")
print("Assistant:", response1.text())

response2 = conversation.prompt("Can you suggest books about that hobby?")
print("Assistant:", response2.text())

# Access conversation history
for response in conversation.responses():
    print(f"Usage: {response.usage()}")

Async Operations

import asyncio
import llm

async def async_example():
    model = llm.get_async_model("gpt-4")
    conversation = model.conversation()
    
    response = await conversation.prompt("Hello, how are you?")
    text = await response.text()
    print(text)
    
    # Async streaming
    response = await conversation.prompt("Tell me a joke", stream=True)
    async for chunk in response:
        print(chunk, end="", flush=True)

asyncio.run(async_example())

Attachments and Media

import llm

model = llm.get_model("gpt-4-vision")

# File attachment
attachment = llm.Attachment(path="/path/to/image.jpg")
response = model.prompt("Describe this image", attachments=[attachment])
print(response.text())

# URL attachment
attachment = llm.Attachment(url="https://example.com/document.pdf")
response = model.prompt("Summarize this document", attachments=[attachment])
print(response.text())

# Binary content
with open("data.png", "rb") as f:
    content = f.read()
attachment = llm.Attachment(content=content, type="image/png")
response = model.prompt("What's in this image?", attachments=[attachment])
print(response.text())

Structured Output

import llm

model = llm.get_model("gpt-4")

# Define JSON schema for structured response
schema = {
    "type": "object",
    "properties": {
        "name": {"type": "string"},
        "age": {"type": "number"},
        "skills": {
            "type": "array",
            "items": {"type": "string"}
        }
    },
    "required": ["name", "age"]
}

response = model.prompt(
    "Create a profile for a software engineer",
    schema=schema
)

# Get structured JSON response
profile = response.response_json()
print(f"Name: {profile['name']}")
print(f"Age: {profile['age']}")
print(f"Skills: {', '.join(profile['skills'])}")

This comprehensive models and conversations system provides the foundation for all LLM interactions in the package, supporting both simple prompting and complex multi-turn conversations with rich media and structured outputs.

Install with Tessl CLI