Overview

Opper is a Unified API that makes it easy to build AI code that is model independent, structured and performant. SDKs are available for Python and Typescript, and they provide a toolkit to perform various common LLM operations.

About this Guide

This guide is optimized for advocating for best practices of building with Opper. It is a good reference for developers, as well as for prompting code editors like Cursor. For a more simple get started, you might want to check out our Get Started guide.

Key concepts

Here are the key concepts of Opper:

A call is a primitive but powerful way of interacting with a generative model. It is a structured definition of a task, with clear input and output schemas where schema annotations are used for prompting. Each call has a "name", which allows it to be managed as a function.
A model is an LLM. A full list of supported LLMs are available at https://docs.opper.ai/capabilities/models.
A span is a log of a call. But unlike ordinary logs, a span is built to be able to hold metrics, evaluations and be tied together with other spans to form what is called a trace. A trace is a chain of spans.
A metric is a data point that you can attach to a span. It is a flexible way to provide feedback on the success of a call from code. It can be used for storing results end user feedback such as thumbs up or down, performance timers or the result of small checks.
A dataset is a collection of example outputs that can act as ground truth for success of a task. Dataset entries can be used for testing changes to prompts, alternative models and be used as examples for the model.
A index is database equipped with semantic similiarty functionallity. You can use it to index knowledge and retrieve the semantically relevant parts of it. Good for memory or knowledge applications.

Best practices

Here are the best practices for building reliable AI implementations with these primitives:

Break up tasks in to steps and implement calls for each step. This helps build an easy to maintain and easy to optimize pipeline.
Always define clear schemas for the output, with descriptions for fields to improve the understanding for the model
Define a schema for the input with descriptions for fields to improve the understanding for the model, and leave out explanation of this in instructions.
Sparingly use natural language instructions, use schemas and field descriptions to define the instructions as much as possible.
Always start a trace to have subsequent calls and indexing operations as children of the parent span. This will make it easier to understand the flow of the application in the Opper UI.
Leave out the model unless you have a specific reason to do so, model optimization is preferably a later optimization.
Perform small and simple evaluations of outputs of calls in code to verify quality or assumptions, to be able to spot issues early
Include a field thoughts as the first item in the output schema to improve quality of the response.

Examples

Get started

We recommend storing your API key for Opper as an environment variable. The SDKs will automatically try to load the key from the environment variable OPPER_API_KEY. You can generate API keys at https://platform.opper.ai

Making calls to models with schemas

In this example we complete a task to extract structured room information from a text. Note how we define input and output schemas for the task and use field descriptions as the primary means to prompt the model.

from opperai import AsyncOpper
from pydantic import BaseModel, Field
import asyncio

opper = AsyncOpper() # Automatically loads api key from environment variable OPPER_API_KEY

# Preferably define a Pydantic model for the input, especially if you have multiple inputs to a call.
class UnstructuredRoomText(BaseModel):
    text: str = Field(description="A text containing information about a hotel room")

# Always define clear Pydantic models for the output
class StructuredRoom(BaseModel):
    thoughts: str = Field(description="The thoughts of the model while extracting the room details") # Recommended to always be present to improve quality of the response
    beds: int = Field(description="The number of people who can sleep in this room")
    seaview: bool = Field(description="Whether the room has a view to the sea")
    description: str = Field(description="A description of the room and its features. Always starting with 'This room features...")

async def main():
    result, _ = await opper.call(
        name="extractRoom", # required, use a unique name for each call/task
        instructions="Extract the room details from the text", # required
        input_type=UnstructuredRoomText,
        output_type=StructuredRoom, # recommended
        model="anthropic/claude-3.7-sonnet", # optional, otherwise the default model will be used
        input=UnstructuredRoomText( text="Suite at Grand Hotel with two rooms. A master bedroom with a king sized bed and a bed sofa for one. The room has a view to the sea, a large bathroom and a balcony."), # required
    )

    print(result)

asyncio.run(main())

import Client from "opperai";

// Initialize the client - automatically loads API key from OPPER_API_KEY environment variable
const client = new Client();

// Define input schema using JSON Schema
const UnstructuredRoomText = {
    type: "object",
    properties: {
        text: {
            type: "string",
            description: "A text containing information about a hotel room",
        },
    },
    required: ["text"],
};

// Define output schema using JSON Schema
const StructuredRoom = {
    type: "object",
    properties: {
        thoughts: {
            type: "string",
            description:
                "The thoughts of the model while extracting the room details", // Recommended to always be present
        },
        beds: {
            type: "number",
            description: "The number of people who can sleep in this room",
        },
        seaview: {
            type: "boolean",
            description: "Whether the room has a view to the sea",
        },
        description: {
            type: "string",
            description:
                "A description of the room and its features. Always starting with 'This room features...'",
        },
    },
    required: ["thoughts", "beds", "seaview", "description"],
};

async function main() {
    const { json_payload } = await client.call({
        name: "extractRoom", // required, use a unique name for each call/task
        instructions: "Extract the room details from the text", // required
        input_schema: UnstructuredRoomText,
        output_schema: StructuredRoom, // recommended
        model: "anthropic/claude-3.7-sonnet", // optional, otherwise the default model will be used
        input: {
            text: "Suite at Grand Hotel with two rooms. A master bedroom with a king sized bed and a bed sofa for one. The room has a view to the sea, a large bathroom and a balcony.",
        },
    });

    console.log(json_payload);
}

main();

Performing tracing and logging

In this example, we implement tracing of a translation task where the an article will be translate to 3 languages by 3 calls. Note how a trace is started as a context manager which makes the individual translations fold into this parent span.

from opperai import AsyncOpper
from pydantic import BaseModel, Field
import asyncio


class ArticleTranslationTask(BaseModel):
    summary: str = Field(description="A concise summary of the main points")
    target_language: str = Field(description="The target language")


class Translation(BaseModel):
    original_language: str = Field(
        description="The original language of the text")
    destination_language: str = Field(
        description="The target language of translation")
    translated_text: str = Field(description="The translated text")


async def main():
    opper = AsyncOpper()

    article = "The rise of artificial intelligence has transformed many industries. Companies are now using AI to automate tasks, improve customer service, and make better decisions. However, this rapid adoption also raises important questions about ethics and job displacement."

    # Start a parent span for this content processing workflow
    async with opper.traces.start("translate_article") as span:

        # We add information on the article to the parent span
        # Note that values have to be strings.
        await span.update(input=str(article))

        target_languages = ["Swedish", "Danish", "Norwegian"]
        translated_texts = []

        for language in target_languages:

            translation_result, _ = await opper.call(
                name="translate",
                instructions="Translate the text",
                input_type=ArticleTranslationTask,
                output_type=Translation,
                input=ArticleTranslationTask(summary=article,
                                             target_language=language),
            )

            translated_texts.append(translation_result.translated_text)

        # If it makes sense, we can also set the output of the session
        await span.update(output="\n".join(translated_texts))


asyncio.run(main())

import Client from "opperai";

const client = new Client();

// Define schemas using JSON Schema
const ArticleTranslationTask = {
    type: "object",
    properties: {
        summary: {
            type: "string",
            description: "A concise summary of the main points",
        },
        target_language: {
            type: "string",
            description: "The target language",
        },
    },
    required: ["summary", "target_language"],
};

const Translation = {
    type: "object",
    properties: {
        original_language: {
            type: "string",
            description: "The original language of the text",
        },
        destination_language: {
            type: "string",
            description: "The target language of translation",
        },
        translated_text: {
            type: "string",
            description: "The translated text",
        },
    },
    required: ["original_language", "destination_language", "translated_text"],
};

async function main() {
    const article =
        "The rise of artificial intelligence has transformed many industries. Companies are now using AI to automate tasks, improve customer service, and make better decisions. However, this rapid adoption also raises important questions about ethics and job displacement.";

    // Start a parent span for this content processing workflow
    const span = await client.traces.start({
        name: "translate_article",
        input: article,
    });

    try {
        const target_languages = ["Swedish", "Danish", "Norwegian"];
        const translated_texts = [];

        for (const language of target_languages) {
            const { json_payload: translation_result } = await client.call({
                name: "translate",
                instructions: "Translate the text",
                input_schema: ArticleTranslationTask,
                output_schema: Translation,
                input: {
                    summary: article,
                    target_language: language,
                },
                parent_span_uuid: span.uuid, // Connect to parent span
            });

            translated_texts.push(translation_result.translated_text);
        }

        await span.end({ output: translated_texts });
    } finally {
    }
}

main();

Performing evals with metrics

In this example, we evaluate accuracy of a task and save a metric. Note how this test is fast and small and tests the main instruction. Metric will be filterable in the UI.

from opperai import AsyncOpper
from pydantic import BaseModel, Field
import asyncio

async def main():
    opper = AsyncOpper()

    result, obj = await opper.call(
        name="answer-question",
        instructions="Answer the question with one word",
        input="What is the capital of France?",
        output_type=str,
    )

    # Perform a simple evaluation
    is_one_word = 1 if len(result.split(" ")) == 1 else 0

    # Save the evaluation to the span
    await obj.span.save_metric(
        dimension="evaluate_is_one_word", 
        value=is_one_word, 
        comment="Evaluated if the answer is one word (1=True, 0=False)"
        )

asyncio.run(main())

import Client from "opperai";

const client = new Client();

async function main() {
  const { message, span_id } = await client.call({
    name: "answer-question",
    instructions: "Answer the question with one word",
    input: "What is the capital of France?",
  });

  // Perform a simple evaluation
  const is_one_word = message.split(" ").length === 1 ? 1 : 0;

  // Save the evaluation to the span
  await client.spans.saveMetric(span_id, {
    dimension: "evaluate_is_one_word",
    value: is_one_word,
    comment: "Evaluated if the answer is one word (1=True, 0=False)"
  });
}

main();

Performing knowledge indexing and retrieval

Here is an example of how to index and retrieve data with the Opper SDK. Note how we pick a descriptive index name, and how we add metadata to documents to be able to use that information when composing responses.

from opperai import AsyncOpper
import asyncio
import time
from pydantic import BaseModel, Field

opper = AsyncOpper()

class Response(BaseModel):
    references: list[str] | None = Field(default=None, description="A list of references to the sources of the information in the response")
    response: str = Field(description="The response to the question with each fact reference as a footnote")

async def main():
    
    index = await opper.indexes.get(name="my-index")

    # We prepare the index if it doesn't exist
    if index is None:
        index = await opper.indexes.create(name="my-index")

        # It takes a while for the index to be ready
        time.sleep(3)

        # Add some text with facts to the index
        await index.add(
            key="document-key",
            content=f"Reddit was founded in 2005. Reddit's headquarters are in San Francisco. Reddit reported a revenue of $100 million in 2023.",
            metadata={
                "source": "https://www.reddit.com",
            },
        )

        # Add another document with Reddit related facts
        await index.add(
            key="additional-fact-document",
            content=f"Reddit has over 430 million monthly active users. Reddit was acquired by Condé Nast in 2006. Reddit's mascot is named Snoo.",
            metadata={
                "source": "https://www.wikipedia.com/reddit",
            },
        )

    query = "What do you know about Reddit?"
    # Query the index with a question related to the facts
    results = await index.query(
        query=query, 
        filters=[{ "key": "source", "operation": "=", "value": "https://www.reddit.com"}], # optional filter
        k=5 # optional number of chunks to return
    )

    # Use the respond function with contents
    response, _ = await opper.call(
        name="answer_question",
        instructions="You are provided with knowledge about a topic. Answer the question using only this provided knowledge. If there is no knowledge, just say you dont know.",
        input={
            "question": query,
            "knowledge": results
        },
        output_type=Response
    )

    print(response)

asyncio.run(main())

import Client from "opperai";

const client = new Client();

// Define response schema using JSON Schema
const Response = {
    type: "object",
    properties: {
        references: {
            type: "array",
            items: {
                type: "string",
            },
            description:
                "A list of references to the sources of the information in the response",
        },
        response: {
            type: "string",
            description:
                "The response to the question with each fact reference as a footnote",
        },
    },
    required: ["response"],
};

async function main() {
    let index = await client.indexes.get("my-index");

    // We prepare the index if it doesn't exist
    if (!index) {
        index = await client.indexes.create("my-index");

        // It takes a while for the index to be ready
        await new Promise((resolve) => setTimeout(resolve, 3000));

        // Add some text with facts to the index
        await index.add({
            key: "document-key",
            content:
                "Reddit was founded in 2005. Reddit's headquarters are in San Francisco. Reddit reported a revenue of $100 million in 2023.",
            metadata: {
                source: "https://www.reddit.com",
            },
        });

        // Add another document with Reddit related facts
        await index.add({
            key: "additional-fact-document",
            content:
                "Reddit has over 430 million monthly active users. Reddit was acquired by Condé Nast in 2006. Reddit's mascot is named Snoo.",
            metadata: {
                source: "https://www.wikipedia.com/reddit",
            },
        });
    }

    const query = "What do you know about Reddit?";
    // Query the index with a question related to the facts
    const results = await index.query({
        query,
        filters: [
            { key: "source", operation: "=", value: "https://www.reddit.com" },
        ], // optional filter
        k: 5, // optional number of chunks to return
    });

    // Use the respond function with contents
    const { json_payload: response } = await client.call({
        name: "answer_question",
        instructions:
            "You are provided with knowledge about a topic. Answer the question using only this provided knowledge. If there is no knowledge, just say you dont know.",
        input: {
            question: query,
            knowledge: results,
        },
        output_schema: Response,
    });

    console.log(response);
}

main();