Validate task outputs with structured result types.

In addition to providing discrete goals for your agents, ControlFlow tasks are designed to translate between the unstructured, conversational world of your AI agents and the structured, programmatic world of your application. The primary mechanism for this translation is the task’s result, which should be a well-defined, validated output that can be used by other tasks or components in your workflow.

ControlFlow allows you to specify the expected structure of a task’s result using the result_type parameter. This ensures that the result conforms to a specific data schema, making it easier to work with and reducing the risk of errors in downstream tasks.

In addition to the basic benefits of type safety and data integrity, result types also serve as a form of documentation for your agents, indicating exactly what kind of data they should expect to produce.

​
Default Results are Strings

By default, the result_type of a task is a string, meaning the agent can return any value that satisfies the task’s objective.

import controlflow as cf

task = cf.Task('Say hello in three languages')
print(task.run())

In the above example, you may get a result like "Hello; Hola; Bonjour", or even something more complex like `“‘Hello!\n\nIn three languages, “Hello” is expressed as follows:\n\n1. English: Hello\n2. Spanish: Hola\n3. French: Bonjour’“.

While this flexibility can be useful in some cases, especially if this task’s result is only being consumed by another ControlFlow task, it can also lead to ambiguity and errors if the agent produces unexpected output.

​
Returning Basic Types

For simple task results, you can use any of Python’s built-in types.

Here’s the above example, specifying that the result should be a list of strings. This guides the agent to give the result you probably expected (three strings, each representing a greeting in a different language):

import controlflow as cf

task = cf.Task('Say hello in three languages', result_type=list[str])
result = task.run()

print(result)
assert isinstance(result, list)
assert len(result) == 3

If your result is a number, you can specify the result_type as int or float:

import controlflow as cf

task = cf.Task("Generate a random number", result_type=int)
result = task.run()

print(result)
assert isinstance(result, int)

You can even use bool for tasks whose result is a simple true/false value:

import controlflow as cf

task = cf.Task("Evaluate the statement: the earth is flat", result_type=bool)
result = task.run()

assert result is False

​
Constrained Choices

Sometimes you want to limit the possible results to a specific set of values. You can do this by specifying a list of allowed values for the result type:

import controlflow as cf

task = cf.Task(
    "Is this a book or a movie?", 
    result_type=["book", "movie"], 
    context=dict(title="Game of Thrones"),
)
result = task.run()

assert result == "book"

​
Pydantic Models

For complex, structured results, you can use a Pydantic model as the result_type. Pydantic models provide a powerful way to define data schemas and validate input data.

import controlflow as cf
from pydantic import BaseModel, Field

class ResearchReport(BaseModel):
    title: str
    summary: str
    key_findings: list[str] = Field(min_items=3, max_items=10)
    references: list[str]

task = cf.Task(
    "Generate a research report on quantum computing",
    result_type=ResearchReport
)

report = task.run()
print(f"Report title: {report.title}")
print(f"Number of key findings: {len(report.key_findings)}")

​
Advanced Validation

Pydantic allows for advanced validation using custom validators:

from pydantic import BaseModel, field_validator

class SentimentAnalysis(BaseModel):
    text: str
    sentiment: float
    
    @field_validator('sentiment')
    def check_sentiment_range(cls, v):
        if not -1 <= v <= 1:
            raise ValueError('Sentiment must be between -1 and 1')
        return v

task = cf.Task(
    "Analyze sentiment of given text",
    result_type=SentimentAnalysis,
    context=dict(text="I love ControlFlow!"),
)

analysis = task.run()
print(f"Sentiment: {analysis.sentiment}")
FlowsCustom Tools