WorkflowAI

https://docs.workflowai.com/
Admin
WorkflowAI is an open-source platform designed for product and engineering teams to collaborate on...
Tokens:11,868
Snippets:75
Trust Score:6.8
Update:1 year ago
Show doc for...
Tokens:
Raw
### Install WorkflowAI Python SDK

Source: https://docs.workflowai.com/python-sdk/get-started

Installs the WorkflowAI Python SDK using pip. Requires Python version 3.9 or higher.

```sh
pip install workflowai
```

--------------------------------

### Run WorkflowAI Agent with OpenAI Model

Source: https://docs.workflowai.com/python-sdk/get-started

Executes the `get_capital_info` agent using the OpenAI GPT-4 Latest model. It prints the structured output and associated cost and latency information.

```python
output = await get_capital_info.run(
    CityInput(city="New York"), 
    model=Model.GPT_4_LATEST
)
print(output)
```

--------------------------------

### Stream WorkflowAI Agent Output with OpenAI

Source: https://docs.workflowai.com/python-sdk/get-started

Streams the output of the `get_capital_info` agent using the OpenAI GPT-4o mini Latest model. The output is printed chunk by chunk as it becomes available.

```python
# use `.stream()` to stream the output
async for chunk in get_capital_info.stream(
    CityInput(city="New York"),
    model=Model.GPT_4O_MINI_LATEST
):
    print(chunk)
```

--------------------------------

### Run WorkflowAI Agent with Gemini Model

Source: https://docs.workflowai.com/python-sdk/get-started

Executes the `get_capital_info` agent using the Gemini 2.0 Flash Latest model. It prints the structured output and associated cost and latency information.

```python
output = await get_capital_info.run(
    CityInput(city="New York"),
    model=Model.GEMINI_2_0_FLASH_LATEST
)
print(output)
```

--------------------------------

### Define a WorkflowAI Agent with Pydantic Models

Source: https://docs.workflowai.com/python-sdk/get-started

Defines an asynchronous agent using the `@workflowai.agent()` decorator. The agent takes a Pydantic model as input and returns a Pydantic model as output. This example defines an agent `get_capital_info` that takes a `CityInput` and returns `CapitalOutput`.

```python
import workflowai
from pydantic import BaseModel
from workflowai import Model

class CityInput(BaseModel):
    city: str

class CapitalOutput(BaseModel):
    country: str
    capital: str 
    fun_fact: str

@workflowai.agent()
async def get_capital_info(city_input: CityInput) -> CapitalOutput:
    ... 
```

--------------------------------

### Initialize WorkflowAI SDK

Source: https://docs.workflowai.com/python-sdk/get-started

Initializes the WorkflowAI SDK with an API key and optional URL. The API key is typically read from the WORKFLOWAI_API_KEY environment variable, and the URL defaults to a global endpoint but can be set to a self-hosted instance via WORKFLOWAI_API_URL.

```python
import os
import workflowai

workflowai.init( # This initialization is optional when using default settings
    api_key=os.environ.get("WORKFLOWAI_API_KEY"),  # This is the default and can be omitted
    url="https://run.workflowai.com",  # This is the default and can be omitted
)
```

--------------------------------

### Run WorkflowAI Agent with Anthropic Model

Source: https://docs.workflowai.com/python-sdk/get-started

Executes the `get_capital_info` agent using the Anthropic Claude 3.5 Sonnet Latest model. It prints the structured output and associated cost and latency information.

```python
output = await get_capital_info.run(
    CityInput(city="New York"),
    model=Model.CLAUDE_3_5_SONNET_LATEST
)
print(output)
```

--------------------------------

### Install Python Packages for Instructor

Source: https://docs.workflowai.com/integrations/instructor

Installs the necessary Python packages: instructor, openai, and pydantic. These are required for using the instructor library with OpenAI or WorkflowAI.

```bash
pip install instructor openai pydantic
```

--------------------------------

### Pydantic Schema with Field Examples

Source: https://docs.workflowai.com/python-sdk/agent

Shows how to provide example outputs for a Pydantic model field. These examples further help in aligning the LLM's behavior and output format.

```python
from pydantic import BaseModel, Field

class Output(BaseModel):
    answer: str = Field(
        description="Answer with bullet points.",
        examples=[
            "- Answer 1",
            "- Answer 2", 
            "- Answer 3"
        ]
    )
```

--------------------------------

### Basic Pydantic Schemas for Agent Input and Output

Source: https://docs.workflowai.com/python-sdk/agent

Illustrates the fundamental Pydantic models for defining an agent's input and output. This example shows a simple question-and-answer structure.

```python
from pydantic import BaseModel

class Input(BaseModel):
    question: str

class Output(BaseModel):
    answer: str
```

--------------------------------

### Install and Use React Feedback Component

Source: https://docs.workflowai.com/features/user-feedback

Instructions for installing the WorkflowAI React SDK and using the FeedbackButtons component in a React application. This component simplifies adding feedback functionality to web apps.

```bash
npm install @workflowai/react
```

```typescript
import { FeedbackButtons } from '@workflowai/react'

...
   <FeedbackButtons feedbackToken={...} userID={...} className='...'/>
...
```

--------------------------------

### List and Check Model Support - Python

Source: https://docs.workflowai.com/python-sdk/agent

This code example shows how to list available models and check their support for specific use cases like audio transcription. It iterates through the models returned by `list_models()` and prints whether each model supports audio based on the `is_not_supported_reason` field.

```python
class AudioInput(BaseModel):
    audio: Audio = Field()

class AudioOutput(BaseModel):
    transcription: str = Field()

@agent(id="audio-transcription")
async def audio_transcription(input: AudioInput) -> AudioOutput:
    """
    Transcribe the audio file.
    """
    ...

models = await audio_transcription.list_models()
for model in models:
    if model.is_not_supported_reason is None:
        print(f"{model.id} supports audio transcription")
    else:
        print(f"{model.id} does not support audio transcription: {model.is_not_supported_reason}")
```

--------------------------------

### Define Agent Instructions with Docstring

Source: https://docs.workflowai.com/python-sdk/agent

Use a Python docstring to provide instructions for the agent. These instructions are passed to the LLM via the system prompt, guiding its behavior and task execution.

```python
@workflowai.agent(id="answer-question")
async def answer_question(input: Input) -> Output:
    """
    You are an expert in history.
    Answer the question with attention to detail and historical accuracy.
    """
    ...
```

--------------------------------

### Template Inputs with Jinja2 in WorkflowAI

Source: https://docs.workflowai.com/integrations/instructor

This Python example demonstrates templating with input variables in WorkflowAI, using Jinja2 syntax for dynamic content injection. It separates static instructions from dynamic data, improving observability and enabling features like benchmarks. Dependencies include 'openai', 'instructor', and 'pydantic'. The function takes an email address and returns its classification.

```python
import os
from typing import Literal

import instructor
from openai import OpenAI
from pydantic import BaseModel


class EmailAddressClassificationOutput(BaseModel):
    kind: Literal["personal", "work", "unsure"]

def classify_email_address(email_address: str) -> EmailAddressClassificationOutput:
    client = instructor.from_openai(
        OpenAI(
            base_url=os.environ["WORKFLOWAI_API_URL"],
            api_key=os.environ["WORKFLOWAI_API_KEY"],
        ),
        mode=instructor.Mode.OPENROUTER_STRUCTURED_OUTPUTS,
    )

    instructions = """You must classify the email address as:
    - 'personal' (gmail, yahoo, etc.),
    - 'work' (company email address)
    - or 'unsure'.
    The email address is:
    {{email_address}}"""

    return client.chat.completions.create(
        model="email-classifier-agent/gpt-4o-mini",
        response_model=EmailAddressClassificationOutput,
        messages=[{"role": "user", "content": instructions}],
        extra_body={"input": {"email_address": email_address}},
    )

if __name__ == "__main__":
    result = classify_email_address("steve@apple.com")
    print(f"Classification: {result.kind}") # 'work'
```

--------------------------------

### Define Synchronous and Asynchronous Python Tools for WorkflowAI

Source: https://docs.workflowai.com/python-sdk/tools

Define custom tools as Python functions, supporting both synchronous and asynchronous operations. These functions require type annotations for schema generation. The example shows a synchronous tool for getting current time and an asynchronous tool for fetching package versions from PyPI.

```python
from datetime import datetime
from zoneinfo import ZoneInfo
import httpx

# Sync tool
def get_current_time(timezone: Annotated[str, "The timezone to get the current time in. e-g Europe/Paris"]) -> str:
    """Return the current time in the given timezone in iso format"""
    return datetime.now(ZoneInfo(timezone)).isoformat()

# Tools can also be async
async def get_latest_pip_version(package_name: Annotated[str, "The name of the pip package to check"]) -> str:
    """Fetch the latest version of a pip package from PyPI"""
    url = f"https://pypi.org/pypi/{package_name}/json"
    async with httpx.AsyncClient() as client:
        response = await client.get(url)
        response.raise_for_status()
        data = response.json()
        return data['info']['version']
```

--------------------------------

### Use @browser-text and @google-search in Python Agent

Source: https://docs.workflowai.com/python-sdk/tools

This Python code snippet demonstrates how to define an agent using WorkflowAI that can utilize the @google-search and @browser-text tools. The agent's docstring specifies the available tools. The example shows how to instantiate and call the agent with a search query.

```python
import workflowai
from workflowai.models import Model

# Assuming SearchWebInput and SearchWebOutput are defined elsewhere

@workflowai.agent(id="web-search-agent", model=Model.CLAUDE_3_5_HAIKU_LATEST)
async def search_web(input: SearchWebInput) -> SearchWebOutput:
    """
    You can use @google-search and @browser-text when relevant.
    """
    ...

# Example usage:
run = await search_web(
    SearchWebInput(
        query="When was the last iPhone released?"
    )
)
print(run)

# Output:

```

--------------------------------

### Code Review Agent with Conditional Instructions

Source: https://docs.workflowai.com/python-sdk/agent

An example agent for code review that uses Jinja2 for conditional logic within its instructions. It adjusts its review strictness based on whether the review is for production or development.

```python
class CodeReviewInput(BaseModel):
    language: str = Field(description="Programming language of the code")
    style_guide: str = Field(description="Style guide to follow")
    is_production: bool = Field(description="Whether this is a production review")
    focus_areas: list[str] = Field(description="Areas to focus on during review", default_factory=list)

class CodeReviewOutput(BaseModel):
    """Output from a code review."""
    issues: list[str] = Field(
        default_factory=list,
        description="List of identified issues or suggestions for improvement"
    )
    compliments: list[str] = Field(
        default_factory=list,
        description="List of positive aspects and good practices found in the code"
    )
    summary: str = Field(
        description="A brief summary of the code review findings"
    )

@workflowai.agent(id="code-review")
async def review_code(review_input: CodeReviewInput) -> CodeReviewOutput:
    """
    You are a code reviewer for {{ language }} code.
    Please review according to the {{ style_guide }} style guide.

    {% if is_production %}
    This is a PRODUCTION review - be extra thorough and strict.
    {% else %}
    This is a development review - focus on maintainability.
    {% endif %}

    {% if focus_areas %}
    Key areas to focus on:
    {% for area in focus_areas %}
    {{ loop.index }}. {{ area }}
    {% endfor %}
    {% endif %}

    Code to review:
    {{ code }}
    """
    ...
```

--------------------------------

### Execute WorkflowAI Agent with Custom Tools (Time Example)

Source: https://docs.workflowai.com/python-sdk/tools

Execute a WorkflowAI agent that utilizes custom tools, demonstrating a query for the current time. The agent will internally call the `get_current_time` tool and return the formatted answer along with execution details like cost and latency.

```python
run = await answer_question(
    AnswerQuestionInput(question="What is the current time in Phoenix, AZ?")
)
print(run)

# Output:
# ==================================================
# {
#   "answer": "The current time in Phoenix, Arizona is 2:42 PM MST (Mountain Standard Time) on February 14, 2025.",
#   "sources": []
# }
# ==================================================
# Cost: $ 0.006414
# Latency: 2.64s
```

--------------------------------

### Extract User Info with Claude 3.7 Sonnet using WorkflowAI

Source: https://docs.workflowai.com/integrations/instructor

This Python code snippet shows how to switch the model to Claude 3.7 Sonnet for extracting user information via WorkflowAI and instructor. It uses the same Pydantic model `UserInfo` and `OPENROUTER_STRUCTURED_OUTPUTS` mode for structured output. Ensure necessary environment variables (`WORKFLOWAI_API_URL`, `WORKFLOWAI_API_KEY`) are set and the required libraries are installed.

```python
import os

import instructor
from openai import OpenAI
from pydantic import BaseModel


class UserInfo(BaseModel):
    name: str
    age: int

def extract_user_info(user_message: str) -> UserInfo:
    client = instructor.from_openai(
        OpenAI(base_url=os.environ["WORKFLOWAI_API_URL"], api_key=os.environ["WORKFLOWAI_API_KEY"]),
        mode=instructor.Mode.OPENROUTER_STRUCTURED_OUTPUTS,
    )

    return client.chat.completions.create(
        model="user-info-extraction-agent/claude-3-7-sonnet-latest", # Agent now runs Claude 3.7 Sonnet
        response_model=UserInfo,
        messages=[{"role": "user", "content": user_message}],
    )

if __name__ == "__main__":
    user_info = extract_user_info("John Black is 33 years old.")
    print("Basic example result:", user_info)  # UserInfo(name='John Black', age=33)
```

--------------------------------

### Reply to a Run in WorkflowAI with Python

Source: https://docs.workflowai.com/python-sdk/agent

Demonstrates how to use the `reply` method of a `Run` object to maintain conversation history in WorkflowAI. It showcases a simple travel chatbot example and how subsequent replies automatically keep track of the conversation context.

```python
class ChatbotInput(BaseModel):
    user_message: str

class Recommendation(BaseModel):
    name: str
    address: str

class ChatbotOutput(BaseModel):
    assistant_message: str
    # You can add structured output to the assistant reply
    recommendations: list[Recommendation]

@workflowai.agent(id="travel-assistant", model=Model.GPT_4O_LATEST)
async def chat(input: ChatbotInput) -> ChatbotOutput:
    """
    A helpful travel assistant that can provide recommendations and answer questions about destinations.
    """
    ...

# Initial question from user
run = await chat.run(ChatbotInput(user_message="I'm planning a trip to Paris. What are the must-see attractions?"))

# Note that the follow-up question does not mention Paris because the conversation history is automatically kept.
reply_run = await run.reply(user_message="When is the best time of year to visit?")
print(reply_run)

# Double-check the answer
confirmation_run = await run.reply(
    user_message="Are you sure?"
)
```

--------------------------------

### Pydantic Schema with Field Description

Source: https://docs.workflowai.com/python-sdk/agent

Demonstrates how to add a description to a Pydantic model field. This description is used to guide the LLM's output generation.

```python
from pydantic import BaseModel, Field

class Output(BaseModel):
    answer: str = Field(description="Answer with bullet points.")
```

--------------------------------

### Execute WorkflowAI Agent with Custom Tools (Package Version Example)

Source: https://docs.workflowai.com/python-sdk/tools

Execute a WorkflowAI agent to retrieve the latest version of a specific package using a custom tool. The agent calls the `get_latest_pip_version` tool and returns the package version along with execution metrics.

```python
run = await answer_question(AnswerQuestionInput(question="What is the latest version of workflowai package?"))
print(run)

# Output:
# ==================================================
# {
#   "answer": "The latest version of the 'workflowai' package is 0.5.5.",
#   "sources": [
#     "PyPI"
#   ]
# }
# ==================================================
# Cost: $ 0.0027
# Latency: 1.38s
```

--------------------------------

### Reference Agent by Version ID (Python)

Source: https://docs.workflowai.com/python-sdk/versions

This Python example shows how to specify a particular agent version to use by providing the `version` argument in the `@workflowai.agent` decorator. This allows targeting a precise minor version, like "2.1".

```python
# this agent will use the version 2.1
@workflowai.agent(id="triage-agent", version="2.1")
```

--------------------------------

### Override Model During Agent Run - Python

Source: https://docs.workflowai.com/python-sdk/agent

This example shows how to override the default model specified in the `@agent` decorator by passing a different model, `Model.CLAUDE_3_5_SONNET_LATEST`, directly to the `run` method. This allows for dynamic model selection at runtime.

```python
run = await answer_question.run(
    Input(question="What is the history of Paris?"),
    model=Model.CLAUDE_3_5_SONNET_LATEST
)
print(run)
```

--------------------------------

### Run Async Generation with WorkflowAI

Source: https://docs.workflowai.com/integrations/instructor

This Python snippet shows how to run generation asynchronously using WorkflowAI, similar to standard OpenAI implementations. It utilizes Pydantic for response models and Instructor for structured outputs. Dependencies include 'openai', 'instructor', and 'asyncio'. It takes user messages as input and returns structured user information.

```python
import os
import asyncio
import instructor
from openai import AsyncOpenAI
from pydantic import BaseModel

class UserInfo(BaseModel):
    name: str
    age: int

async def extract_user_info_async(user_message: str) -> UserInfo:
    client = instructor.from_openai(
        AsyncOpenAI(base_url=os.environ["WORKFLOWAI_API_URL"], api_key=os.environ["WORKFLOWAI_API_KEY"]),
        mode=instructor.Mode.OPENROUTER_STRUCTURED_OUTPUTS,
    )

    return await client.chat.completions.create(
        model="user-info-extraction-agent/claude-3-7-sonnet-latest",
        response_model=UserInfo,
        messages=[{"role": "user", "content": user_message}],
    )
    
if __name__ == "__main__":
    user_info = asyncio.run(extract_user_info_async("John Black is 33 years old."))
    print("Basic example result:", user_info)  # UserInfo(name='John Black', age=33)
```

--------------------------------

### Running Agent with Dynamic Instructions

Source: https://docs.workflowai.com/python-sdk/agent

Shows how to run an agent that utilizes Jinja2 templated instructions. The `Input` model's fields are used to populate the template variables, creating a specific system prompt for the run.

```python
# Run the agent
run = await answer_question.run(
    Input(
        question="What is artificial intelligence?",
        word_count=5
    )
)

# View prompt
# https://workflowai.com/docs/agents/answer-question-with-word-count/runs/019509ed-017e-7059-4c25-6137ebdb7dcd
# System prompt:
# <instructions>The answer should be less than 5 words. Answer the following question: What is artificial intelligence?</instructions>
# { "answer": "Smart computer systems learning" }
```

--------------------------------

### Configure WorkflowAI Credentials for Instructor

Source: https://docs.workflowai.com/integrations/instructor

Sets the WorkflowAI API key and URL as environment variables. This allows the Instructor library to authenticate with WorkflowAI and use its services.

```bash
export WORKFLOWAI_API_KEY=<your-workflowai-api-key>
export WORKFLOWAI_API_URL=https://run.workflowai.com/v1
```

--------------------------------

### Configure Multiple WorkflowAI Clients in Python

Source: https://docs.workflowai.com/python-sdk/agent

Illustrates how to create and use multiple WorkflowAI client instances with different API keys or accounts. This allows for more flexible management of WorkflowAI agents and resources.

```python
from workflowai import WorkflowAI

client = WorkflowAI(
    url=...,
    api_key=...,
)

# Use the client to create and run agents
@client.agent()
def my_agent(agent_input: Input) -> Output:
    ...
```

--------------------------------

### Extract User Info with GPT-4o-mini using WorkflowAI

Source: https://docs.workflowai.com/integrations/instructor

This Python code snippet demonstrates how to extract user information (name and age) from a given message using the WorkflowAI API and the instructor library. It leverages the `OPENROUTER_STRUCTURED_OUTPUTS` mode for guaranteed structured output. The output is a Pydantic model `UserInfo`. It requires `instructor`, `openai`, and `pydantic` libraries, and environment variables `WORKFLOWAI_API_URL` and `WORKFLOWAI_API_KEY`.

```python
import os

import instructor
from openai import OpenAI
from pydantic import BaseModel


class UserInfo(BaseModel):
    name: str
    age: int


def extract_user_info(user_message: str) -> UserInfo:
    client = instructor.from_openai(
        OpenAI(
            base_url=os.environ["WORKFLOWAI_API_URL"], # OpenAI now uses WorkflowAI's URL and API key
            api_key=os.environ["WORKFLOWAI_API_KEY"], # Get your API key with $5 free credits at workflowai.com/developers/python/instructor
        ),
        mode=instructor.Mode.OPENROUTER_STRUCTURED_OUTPUTS,
    )

    return client.chat.completions.create(
        model="user-info-extraction-agent/gpt-4o-mini-latest",  # Recommendation: use '<agent_name>/<model_name>' format, see why in the next section. 
        response_model=UserInfo,
        messages=[{"role": "user", "content": user_message}],
    )


if __name__ == "__main__":
    user_info = extract_user_info("John Doe is 32 years old.")
    print("Basic example result:", user_info)  # UserInfo(name='John Doe', age=32)
```

--------------------------------

### Post Feedback using TypeScript SDK

Source: https://docs.workflowai.com/features/user-feedback

Shows how to post feedback using the WorkflowAI TypeScript SDK. It includes initializing the SDK client and calling the sendFeedback method with necessary parameters.

```typescript
import { WorkflowAI } from "@workflowai/workflowai";

const workflowAI = WorkflowAI()

await workflowAI.sendFeeback({feedback_token: "", outcome: "positive", comment: "...", userID: ""})
```

--------------------------------

### Run Agent with Input - Python

Source: https://docs.workflowai.com/python-sdk/agent

This snippet illustrates how to execute an agent, `answer_question`, with a specific input. It calls the `run` method with an `Input` object containing a question and prints the result, demonstrating the agent's execution and output.

```python
run = await answer_question.run(Input(question="What is the history of Paris?"))
print(run)
```

--------------------------------

### Add WorkflowAI Credentials to .env File

Source: https://docs.workflowai.com/integrations/instructor

Stores WorkflowAI API key and URL in a .env file for environment variable loading. This is an alternative to setting environment variables directly, useful for local development.

```bash
WORKFLOWAI_API_KEY=<your-workflowai-api-key>
WORKFLOWAI_API_URL=https://run.workflowai.com/v1
```

--------------------------------

### System Prompt Generation from Docstring

Source: https://docs.workflowai.com/python-sdk/agent

Illustrates how a Python docstring is transformed into a system prompt for the LLM. This conversion makes the agent's instructions explicit and usable by the language model.

```python
system_prompt = """<instructions>You are an expert in history. Answer the question with attention to detail and historical accuracy.</instructions>"""
```

--------------------------------

### Agent with Jinja2 Variables in Instructions

Source: https://docs.workflowai.com/python-sdk/agent

Demonstrates using Jinja2 templating within agent docstrings to dynamically inject variables from input models. This allows for customizable instructions based on runtime data.

```python
class Input(BaseModel):
    question: str
    word_count: int

class Output(BaseModel):
    answer: str

@workflowai.agent(id="answer-question-with-word-count", model=Model.CLAUDE_3_5_HAIKU_LATEST)
async def answer_question(input: Input) -> Output:
    """
    The answer should be less than {{ word_count }} words.
    Answer the following question:
    {{ question }}
    """
    ...
```

--------------------------------

### Process Local PDF File with WorkflowAI Agent

Source: https://docs.workflowai.com/python-sdk/multimodality

This code snippet demonstrates how to read a local PDF file, create a `PDF` object, and use it as input to an agent for question answering. It involves opening the file in binary read mode and passing the content to the `PDF` class.

```python
pdf_path = "path/to/pdf.pdf"
with open(pdf_path, "rb") as pdf_file:
    content = pdf_file.read()

pdf = PDF(content_type="application/pdf", data=content)
agent_run = await answer_pdf_question.run(
    PDFQuestionInput(
        pdf=pdf,
        question="What are the main points from the document?"
    )
)
print(agent_run)
```

--------------------------------

### Answer Questions from PDF Agent (Python)

Source: https://docs.workflowai.com/python-sdk/multimodality

This Python code defines an agent capable of answering questions based on a PDF document. It utilizes the `workflowai` library and Pydantic for input and output structuring. The agent takes a PDF file (via URL) and a question as input, returning an answer and supporting quotes. It is optimized for PDF processing using the `GEMINI_2_0_FLASH_LATEST` model.

```python
import workflowai
from pydantic import BaseModel, Field
from workflowai import Model
from workflowai.fields import PDF

class PDFQuestionInput(BaseModel):
    pdf: PDF = Field(description="The PDF document to analyze")
    question: str = Field(description="The question to answer about the PDF content")

class PDFAnswerOutput(BaseModel):
    answer: str = Field(description="The answer to the question based on the PDF content")
    quotes: list[str] = Field(description="Relevant quotes from the PDF that support the answer")

@workflowai.agent(id="pdf-answer-bot", model=Model.GEMINI_2_0_FLASH_LATEST)
async def answer_pdf_question(input: PDFQuestionInput) -> PDFAnswerOutput:
    """
    Analyze the provided PDF document and answer the given question.
    Provide a clear and concise answer based on the content found in the PDF.
    Include relevant quotes to support the answer.
    """
    ...

# Run the agent

## From a remote file
pdf_url = "https://microsoft.gcs-web.com/static-files/b3eef820-6757-44ea-9f98-3963bace4837"
pdf = PDF(url=pdf_url)
agent_run = await answer_pdf_question.run(
    PDFQuestionInput(
        pdf=pdf,
        question="What are the main points from the document?"
    )
)
print(agent_run)
```

--------------------------------

### Fetch Raw LLM Completions with Python

Source: https://docs.workflowai.com/python-sdk/agent

Shows how to retrieve the exact prompt sent to an AI provider and the raw response received using the `fetch_completions` method on a run object in Python. This is useful for debugging, analyzing token usage, and auditing model interactions. The method returns a list of completion objects, which can be serialized to JSON.

```python
# Fetch the raw completion from the LLM
run = await answer_question.run(Input(question="What is the history of Paris?"))

# Get completion details
completions = await run.fetch_completions()

for completion in completions:
    completion_json = completion.model_dump_json(indent=2)
    print(completion_json)
```

--------------------------------

### Run WorkflowAI Agent with Local Audio File

Source: https://docs.workflowai.com/python-sdk/multimodality

This Python code illustrates how to run a WorkflowAI agent with a local audio file. It reads the audio file, creates an `Audio` object, and passes it within the `AudioInput` model to the agent's `run` method, specifying a particular model.

```python
# Run the agent
audio_path = "path/to/audio.mp3"
with open(audio_path, "rb") as audio_file:
    content = audio_file.read()

audio = Audio(content_type="audio/mp3", data=content)
agent_run = await detect_audio_spam.run(
    AudioInput(audio=audio),
    model=Model.GEMINI_2_0_FLASH_LATEST
)
print(agent_run)
```

--------------------------------

### Post Feedback using Python SDK

Source: https://docs.workflowai.com/features/user-feedback

Demonstrates how to send feedback using the WorkflowAI Python SDK. This function allows for specifying the feedback token, outcome, comment, and user ID.

```python
import workflowai

await workflowai.send_feedback(feedback_token="...", outcome="positive", comment=..., user_id=...)
```

--------------------------------

### Define PDF Question Answering Schemas in Python

Source: https://docs.workflowai.com/concepts/schemas

Demonstrates how to define input and output schemas for a PDF question-answering task using Pydantic in Python. It includes type hints for file uploads, text questions, and structured output with supporting quotes.

```python
from typing import Optional, List
from pydantic import BaseModel

class PdfQuestionAnsweringTaskInput(BaseModel):
    pdf_document: Optional[File] = None
    question: Optional[str] = None

class SupportingQuote(BaseModel):
    quote: Optional[str] = None
    page_number: Optional[float] = None

class PdfQuestionAnsweringTaskOutput(BaseModel):
    answer: Optional[str] = None
    supporting_quotes: Optional[List[SupportingQuote]] = None
```

--------------------------------

### Python Image Generation with WorkflowAI

Source: https://docs.workflowai.com/use-cases/image-generation

This Python code demonstrates how to define and run an image generation feature using the WorkflowAI SDK. It includes defining input and output models, setting up an agent with a specific model, and running the agent to generate an image. The output image's content type and base64 encoded data can be accessed.

```python
from pydantic import BaseModel

import workflowai
from workflowai.fields import Image


class ImageGenerationInput(BaseModel):
    prompt: str


class ImageGenerationOutput(BaseModel):
    image: Image


@workflowai.agent(id="image-generation", model="gpt-image-1")
async def image_generation(_: ImageGenerationInput) -> ImageGenerationOutput: ...


async def generate_image(prompt: str):
    run = await image_generation.run(ImageGenerationInput(prompt=prompt))
    image = run.output.image
    print(image.content_type)
    print(image.data)  # data is the base64 encoded image data
```

--------------------------------

### Classify Email Address with Python Deployment

Source: https://docs.workflowai.com/integrations/instructor

This Python function demonstrates how to use WorkflowAI Deployments to classify an email address. It utilizes the instructor library and OpenAI client, setting the model to a specific deployment ID. Input variables are passed via `extra_body['input']` as messages are not required.

```python
import os
from typing import Literal

import instructor
from openai import OpenAI
from pydantic import BaseModel


class EmailAddressClassificationOutput(BaseModel):
    kind: Literal["personal", "work", "unsure"]

def classify_email_address_deployment(email_address: str) -> EmailAddressClassificationOutput:
    client = instructor.from_openai(
        OpenAI(
            base_url=os.environ["WORKFLOWAI_API_URL"],
            api_key=os.environ["WORKFLOWAI_API_KEY"],
        ),
        mode=instructor.Mode.OPENROUTER_STRUCTURED_OUTPUTS,
    )

    return client.chat.completions.create(
        model="email-classifier-agent/#1/production",
        response_model=EmailAddressClassificationOutput,
        messages=[],  # No messages needed; instructions come from the deployment
        extra_body={"input": {"email_address": email_address}},
    )

if __name__ == "__main__":
    result = classify_email_address_deployment("john.doe@gmail.com")
    print(f"Deployment classification: {result.kind}") # 'personal'
```

--------------------------------

### Run Task API Endpoint

Source: https://docs.workflowai.com/features/user-feedback

This section details the API endpoint for running a task. It specifies the HTTP method, host, required headers, and the structure of the JSON request body and response.

```shell
POST /v1/_/agents/my-agent/schemas/1/run
Host: https://run.workflowai.com
Authorization: Bearer {Add your API key here}
Content-Type: application/json

# JSON Body
{
   "task_input": ...
}

# Response
{
   "task_output": ...,
   "feedback_token": ...
}
```

--------------------------------

### Access Cost and Latency - Python

Source: https://docs.workflowai.com/python-sdk/agent

This code demonstrates how to retrieve and print the cost and latency information associated with an agent's run. It accesses these details from the `run` object using `run.cost_usd` and `run.duration_seconds` attributes.

```python
run = await answer_question.run(Input(question="What is the history of Paris?"))
print(f"Cost: $ {run.cost_usd:.5f}")
print(f"Latency: {run.duration_seconds:.2f}s")
```

--------------------------------

### List Supported Models for Audio Input in WorkflowAI

Source: https://docs.workflowai.com/python-sdk/multimodality

This Python code demonstrates how to retrieve a list of models available for a WorkflowAI agent and filter them to find models that support audio input. It iterates through the model list and checks the `is_not_supported_reason` field.

```python
models = await detect_audio_spam.list_models()
# Filter for supported models
supported_models = [model.id for model in models if model.is_not_supported_reason is None]
# Now supported_models contains all models that can process audio
```

--------------------------------

### Create Initial Triage Agent (Python)

Source: https://docs.workflowai.com/python-sdk/versions

This Python code defines an agent for triaging customer questions into predefined categories using the WorkflowAI library. It sets up input and output models and an agent function. Running this agent for the first time automatically creates a new version.

```python
from typing import Literal
from pydantic import BaseModel
import workflowai

class Input(BaseModel):
    question: str

class Output(BaseModel):
    category: Literal["billing", "technical", "account", "other"]

@workflowai.agent(id="triage-agent")
async def triage_question(input: Input) -> Output:
    """
    Triage a customer question into different categories.
    """
    ...

await triage_question.run(Input(question="How do I change my billing information?"))
```

--------------------------------

### Define PDF Question Answering Schemas in TypeScript

Source: https://docs.workflowai.com/concepts/schemas

Illustrates the definition of input and output interfaces for a PDF question-answering task in TypeScript. This includes optional file and string inputs, and structured output with an array of supporting quotes.

```typescript
interface PdfQuestionAnsweringTaskInput {
    pdf_document?: File;
    question?: string;
}

interface SupportingQuote {
    quote?: string;
    page_number?: number;
}

interface PdfQuestionAnsweringTaskOutput {
    answer?: string;
    supporting_quotes?: SupportingQuote[];
}
```

--------------------------------

### Access Feedback Token with Python SDK

Source: https://docs.workflowai.com/features/user-feedback

Demonstrates how to retrieve the feedback token from an AI agent's run or stream operation using the WorkflowAI Python SDK. The token is essential for associating user feedback with specific AI interactions.

```python
import workflowai

# Get feedback token from run
run = await my_agent.run(MyAgentInput())
print(run.feedback_token)

# Get feedback token when streaming
async for chunk in my_agent.stream(MyAgentInput()):
    # Process chunks
    pass
print(chunk.feedback_token)
```

--------------------------------

### WorkflowAI Agent Definition with Pydantic Schemas

Source: https://docs.workflowai.com/python-sdk/agent

Defines an agent for processing customer feedback using Pydantic models for structured input and output. The agent is decorated with @workflowai.agent and specifies the LLM model to use.

```python
class FeedbackInput(BaseModel):
    reviews: list[str]

class FeedbackOutput(BaseModel):
    positive_aspects: list[str]
    negative_aspects: list[str]

@workflowai.agent(id="feedback")
async def feedback(input: FeedbackInput) -> FeedbackOutput:
    """
    Identify what are the positive and negative aspects.
    """
    ...

run = await feedback.run(
    FeedbackInput(reviews=["..."]),
    model=Model.GPT_4O_LATEST
)
```

--------------------------------

### Stream AI Output with Python

Source: https://docs.workflowai.com/python-sdk/agent

Demonstrates how to stream AI model output using the `stream` method in Python. This method returns an AsyncIterator, allowing for processing of output chunks as they become available within an async for loop. Partial outputs are guaranteed to be valid schemas.

```python
async for chunk in answer_question.stream(Input(question="What is the history of Paris?')):
    print(chunk)
```

--------------------------------

### Create WorkflowAI Agent for Audio Spam Detection

Source: https://docs.workflowai.com/python-sdk/multimodality

This Python snippet shows the definition of an asynchronous agent, `detect_audio_spam`, designed to detect spam in audio files. It uses the `AudioInput` model for input and is expected to return an `AudioOutput` model indicating whether the audio is spam.

```python
class AudioOutput(BaseModel):
    is_spam: bool = Field(description="Whether the audio contains spam")
    explanation: str = Field(description="Explanation of the result")

@workflowai.agent(id="audio-spam-detector")
async def detect_audio_spam(input: AudioInput) -> AudioOutput:
    """
    Analyze the provided audio file and determine if it contains spam.
    """
    ...
```

--------------------------------

### Update Agent Instructions for Major Version (Python)

Source: https://docs.workflowai.com/python-sdk/versions

This Python code illustrates how to create a major version by modifying the agent's instructions within the docstring. Changes to instructions, temperature, descriptions, or tools trigger a new major version.

```python
async def triage_question(input: Input) -> Output:
    """
    Triage a customer question into different categories.

    Categories:
    - billing: Questions about payments, invoices, pricing, or subscription changes
    - technical: Questions about API usage, SDK implementation, or technical issues
    - account: Questions about account access, settings, or profile management
    - other: Questions that don't fit into the above categories
    """
    ...
```

--------------------------------

### Run Task API

Source: https://docs.workflowai.com/features/user-feedback

This endpoint allows you to run a task within the WorkflowAI system. It returns a feedback token that can be used to submit feedback on the task's outcome.

```APIDOC
## POST /v1/_/agents/my-agent/schemas/1/run

### Description
Executes a specified task within the WorkflowAI system.

### Method
POST

### Endpoint
`/v1/_/agents/{agent_id}/schemas/{schema_id}/run`

### Parameters
#### Path Parameters
- **agent_id** (string) - Required - The ID of the agent.
- **schema_id** (string) - Required - The ID of the schema for the task.

#### Request Body
- **task_input** (any) - Required - The input payload for the task.

### Request Example
```json
{
   "task_input": { ... }
}
```

### Response
#### Success Response (200)
- **task_output** (any) - The output of the executed task.
- **feedback_token** (string) - A token to be used for submitting feedback.
```

--------------------------------

### Handle WorkflowAIError in Python

Source: https://docs.workflowai.com/python-sdk/errors

Demonstrates how to define input and output models, create a WorkflowAI agent, and handle WorkflowAIError exceptions when calling the agent. This includes accessing the error code and message.

```python
from datetime import date
import workflowai
from pydantic import BaseModel, Field
from workflowai import Model, WorkflowAIError

# define your input and output fields
class Input(BaseModel):
    transcript: str
    call_date: date

class Output(BaseModel):
    positive_points: list[str] = Field(description="List of positive points from the call", default_factory=list)
    negative_points: list[str] = Field(description="List of negative points from the call", default_factory=list)

# define your agent
@workflowai.agent(model=Model.GEMINI_2_0_FLASH_LATEST)
async def analyze_call_feedback(input: Input) -> Output:
    """
    Analyze customer call feedback and extract positive and negative points.
    """
    ...

try:
    await analyze_call_feedback(
        CallFeedbackInput(
            transcript="[00:01:15] Customer: The product is great!",
            call_date=date(2024, 1, 15)
        )
    )
except WorkflowAIError as e:
    print(e.code)
    print(e.message)
```

--------------------------------

### Define Feedback Analysis Schemas (Python)

Source: https://docs.workflowai.com/python-sdk/schemas

Defines Pydantic models for analyzing customer feedback calls, including input for the transcript and call date, a model for individual feedback points with quotes, and an output model to structure positive and negative feedback. This enables systematic analysis of call transcripts.

```python
class FeedbackInput(BaseModel):
    """Input for analyzing a customer feedback call."""
    transcript: str = Field(description="The full transcript of the customer feedback call.")
    call_date: date = Field(description="The date when the call took place.")

# Model representing a single feedback point with supporting evidence
class FeedbackPoint(BaseModel):
    """A specific feedback point with its supporting quote."""
    point: str = Field(description="The main point or insight from the feedback.")
    quote: str = Field(description="The exact quote from the transcript supporting this point.")
    timestamp: str = Field(description="The timestamp or context of when this was mentioned in the call.")

# Model representing the structured analysis of the customer feedback call
class FeedbackOutput(BaseModel):
    """Structured analysis of the customer feedback call."""
    positive_points: list[FeedbackPoint] = Field(
        default_factory=list,
        description="List of positive feedback points, each with a supporting quote."
    )
    negative_points: list[FeedbackPoint] = Field(
        default_factory=list,
        description="List of negative feedback points, each with a supporting quote."
    )
```

--------------------------------

### Define Audio Input for WorkflowAI Agent

Source: https://docs.workflowai.com/python-sdk/multimodality

This Python code defines a Pydantic model `AudioInput` that includes an `Audio` field. This model is used to structure the input for agents that process audio data, leveraging the `workflowai.fields.Audio` class.

```python
from workflowai.fields import Audio

class AudioInput(BaseModel):
    audio: Audio = Field()
```

--------------------------------

### Access Feedback Token with TypeScript SDK

Source: https://docs.workflowai.com/features/user-feedback

Illustrates how to obtain the feedback token when running or streaming with an AI agent using the WorkflowAI TypeScript SDK. This token is crucial for linking user feedback to the corresponding AI interaction.

```typescript
import { WorkflowAI } from "@workflowai/workflowai";

const workflowAI = WorkflowAI();

// Get feedback token from run
const { output, feedbackToken } = await myAgentFunction(input);
console.log(feedbackToken);

// Get feedback token when streaming
let lastChunk: RunStreamEvent<OutputType> | undefined;
for await (const chunk of stream) {
    lastChunk = chunk;
}
console.log(lastChunk?.feedbackToken);
```

--------------------------------

### Define WorkflowAI Agent for Production Deployment

Source: https://docs.workflowai.com/python-sdk/versions

This snippet shows how to use the @workflowai.agent decorator to define an agent specifically for a production deployment. It requires the `id` and `deployment` parameters to be set.

```python
@workflowai.agent(id="triage-agent", deployment="production") # or simply @workflowai.agent()
```

--------------------------------

### Define Audio Input and Spam Classification Models (Python)

Source: https://docs.workflowai.com/python-sdk/schemas

These Pydantic models define the structure for audio input and the expected output for spam classification. AudioInput specifies the audio data, SpamIndicator details specific spam traits with quotes, and AudioClassification provides the spam verdict, confidence, indicators, and reasoning. These are crucial for processing and analyzing audio content.

```python
class AudioInput(BaseModel):
    """Input containing the audio file to analyze."""
    audio: Audio = Field(
        description="The audio recording to analyze for spam/robocall detection",
    )


class SpamIndicator(BaseModel):
    """A specific indicator that suggests the call might be spam."""
    description: str = Field(
        description="Description of the spam indicator found in the audio",
        examples=[
            "Uses urgency to pressure the listener",
            "Mentions winning a prize without entering a contest",
            "Automated/robotic voice detected",
        ],
    )
    quote: str = Field(
        description="The exact quote or timestamp where this indicator appears",
        examples=[
            "'You must act now before it's too late'",
            "'You've been selected as our prize winner'",
            "0:05-0:15 - Synthetic voice pattern detected",
        ],
    )


class AudioClassification(BaseModel):
    """Output containing the spam classification results."""
    is_spam: bool = Field(
        description="Whether the audio is classified as spam/robocall",
    )
    confidence_score: float = Field(
        description="Confidence score for the classification (0.0 to 1.0)",
        ge=0.0,
        le=1.0,
    )
    spam_indicators: list[SpamIndicator] = Field(
        default_factory=list,
        description="List of specific indicators that suggest this is spam",
    )
    reasoning: str = Field(
        description="Detailed explanation of why this was classified as spam or legitimate",
    )
```

--------------------------------

### Define WorkflowAI Agent for Staging Deployment

Source: https://docs.workflowai.com/python-sdk/versions

This snippet demonstrates how to configure a WorkflowAI agent for a staging environment. The `@workflowai.agent` decorator is applied with `id` and `deployment` parameters specifying 'staging'.

```python
@workflowai.agent(id="triage-agent", deployment="staging")
```

--------------------------------

### Define Image Input and Output Schemas (Python)

Source: https://docs.workflowai.com/python-sdk/schemas

Defines Pydantic models for image analysis, specifying an input model expecting an image object and an output model for extracting city, country, and confidence level of identification. This schema is used for analyzing visual content.

```python
class ImageInput(BaseModel):
    image: Image = Field(description="The image to analyze")


class ImageOutput(BaseModel):
    city: str = Field(default="", description="Name of the city shown in the image")
    country: str = Field(default="", description="Name of the country where the city is located")
    confidence: Optional[float] = Field(
        default=None,
        description="Confidence level in the identification (0-1)",
    )
```

--------------------------------

### Identify City from Image Agent (Python)

Source: https://docs.workflowai.com/python-sdk/multimodality

This Python code defines an agent that identifies the city and country from an image. It uses the `workflowai` library and Pydantic for input/output validation. The agent can process images from URLs or base64 encoded local files. Dependencies include `workflowai`, `pydantic`, and `base64`.

```python
import base64
import workflowai
from workflowai import Model
from workflowai.fields import Image
from pydantic import BaseModel, Field
from typing import Optional

class ImageInput(BaseModel):
    image: Image = Field(description="The image to analyze")

class ImageOutput(BaseModel):
    city: str = Field(default="", description="Name of the city shown in the image")
    country: str = Field(default="", description="Name of the country where the city is located")
    confidence: Optional[float] = Field(
        default=None,
        description="Confidence level in the identification (0-1)",
    )

@workflowai.agent(id="city-identifier", model=Model.GEMINI_2_0_FLASH_LATEST)
async def identify_city_from_image(image_input: ImageInput) -> ImageOutput:
    """
    Analyze the provided image and identify the city and country shown in it.
    If the image shows a recognizable landmark or cityscape, identify the city and country.
    If uncertain, indicate lower confidence or leave fields empty.

    Focus on:
    - Famous landmarks
    - Distinctive architecture
    - Recognizable skylines
    - Cultural elements that identify the location

    Return empty strings if the city/country cannot be determined with reasonable confidence.
    """
    ...

# Run the agent

## From a remote image
image_url = "https://images.pexels.com/photos/699466/pexels-photo-699466.jpeg"
image = Image(url=image_url)
agent_run = await identify_city_from_image.run(ImageInput(image=image))
print(agent_run)

## From a local image
image_path = "path/to/image.jpg"
with open(image_path, "rb") as image_file:
    content = base64.b64encode(image_file.read()).decode("utf-8")

image = Image(content_type="image/jpeg", data=content)
agent_run = await identify_city_from_image.run(ImageInput(image=image))
```

--------------------------------

### Handle Recoverable Errors with Partial Output in Python

Source: https://docs.workflowai.com/python-sdk/errors

Shows how an agent returning a full run object can recover from errors, providing both an error and a partial output in the run object. This is useful when dealing with partially valid LLM outputs.

```python
run = await agent(input=Input(name="John"))

# The run will have an error
assert run.error is not None

# The run will have a partial output
assert run.output is not None
```
Tokens:
Raw