Task-Centric Structure

Overview

In the Upsonic framework, a Task is a specific assignment completed by an Agent. Tasks provide all necessary details for execution, such as a description, tools, response format, caching options, and more, facilitating a wide range of action complexities. Tasks within Upsonic can be collaborative, requiring multiple agents to work together through context sharing and task dependencies. Tasks are executed by agents in a single-threaded manner, with each task processed sequentially. The execution flow includes task validation, context processing, tool processing, agent execution, response processing, and caching.

Task Attributes

The Task class provides comprehensive configuration options to customize task behavior and execution.

Core Attributes

Attribute	Type	Description
description	str	A clear, concise statement of what the task entails
tools	List[Any]	The tools/resources the agent can use for this task
response_format	Union[Type[BaseModel], type[str], None]	The expected output format (string, Pydantic model, or None)
context	Any	All context for this task - can include other tasks, files, images, documents, knowledge bases, or any other contextual information

Advanced Configuration

Attribute	Type	Description
enable_thinking_tool	Optional[bool]	Whether to enable the thinking tool for complex reasoning
enable_reasoning_tool	Optional[bool]	Whether to enable the reasoning tool for multi-step analysis
guardrail	Optional[Callable]	Function to validate task output before proceeding
guardrail_retries	Optional[int]	Maximum number of retries when guardrail validation fails
enable_cache	bool	Whether to enable caching for this task
cache_method	Literal[“vector_search”, “llm_call”]	Method to use for caching
cache_threshold	float	Similarity threshold for cache hits (0.0-1.0)
cache_duration_minutes	int	How long to cache results in minutes

Creating a Task

Tasks in Upsonic are created directly in code using the Task class constructor. Each task can be customized with specific tools, response formats, caching options, and validation rules to meet your exact requirements.

Basic Task Creation

from upsonic import Task

# Simple task with just a description
task = Task(description="What is the capital of France?")

Task with Tools

from upsonic import Task
from upsonic.tools import tool

# Create a simple tool
@tool
def calculator(operation: str, a: float, b: float) -> str:
    """Perform basic mathematical operations."""
    if operation == "add":
        return f"Result: {a + b}"
    elif operation == "multiply":
        return f"Result: {a * b}"
    return "Invalid operation"

# Create task with tools
task = Task(
    description="Calculate 10 + 5 using the calculator tool",
    tools=[calculator]
)

Task with Response Format

from pydantic import BaseModel

class AnalysisResult(BaseModel):
    summary: str
    confidence: float
    recommendations: list[str]

# Task with structured response
task = Task(
    description="Analyze the provided data and provide structured results",
    response_format=AnalysisResult
)

Adding Tools to a Task

Tasks can be equipped with various tools to extend their capabilities. Tools are added through the tools parameter and can include custom functions, built-in tools, or tool collections.

Single Tool

from upsonic.tools import tool

@tool
def web_search(query: str) -> str:
    """Search the web for information."""
    return f"Search results for: {query}"

task = Task(
    description="Search for information about AI",
    tools=[web_search]
)

Multiple Tools

from upsonic.tools import tool

@tool
def web_search(query: str) -> str:
    """Search the web for information."""
    return f"Search results for: {query}"

@tool
def data_processor(data: str) -> str:
    """Process and analyze data."""
    return f"Processed: {data}"

# Task with multiple tools
task = Task(
    description="Search for information about AI and process the results",
    tools=[web_search, data_processor]
)

Tool Collections

from upsonic.tools import YFinanceTools

# Create a tool collection
finance_tools = YFinanceTools(
    stock_price=True,
    company_info=True,
    analyst_recommendations=True
)

# Task with tool collection
task = Task(
    description="Get stock information for AAPL",
    tools=finance_tools.functions()
)

Tool Configuration

from upsonic.tools import tool

@tool(requires_confirmation=True, cache_results=True)
def sensitive_operation(data: str) -> str:
    """Perform a sensitive operation that requires confirmation."""
    return f"Processed sensitive data: {data}"

task = Task(
    description="Process sensitive information",
    tools=[sensitive_operation]
)

Adding Tasks to Other Tasks in Context

In Upsonic, you can use the output of one task as context for another task through the context parameter. This enables task chaining and complex workflows.

Basic Task Chaining

# First task
task1 = Task(description="What is 2 + 2?")
result1 = agent.do(task1)

# Second task using context from first task
task2 = Task(
    description="Based on the previous result, what is that number multiplied by 3?",
    context=[task1]
)
result2 = agent.do(task2)

Multiple Context Sources

# Task with multiple context sources
task3 = Task(
    description="Summarize all the previous mathematical results",
    context=[task1, task2, "Additional context: These were all math problems"]
)
result3 = agent.do(task3)

Complex Workflow Example

# Task 1: Data collection
task1 = Task(
    description="Collect data from market research sources",
    tools=[data_collector],
    enable_cache=True
)

# Task 2: Data analysis
task2 = Task(
    description="Analyze the collected market data",
    tools=[data_analyzer],
    context=[task1],
    enable_thinking_tool=True
)

# Task 3: Report generation
task3 = Task(
    description="Generate a comprehensive market report",
    context=[task1, task2],
    response_format=ReportResult
)

Putting Knowledge Base to Task

Knowledge bases can be added to tasks to provide RAG (Retrieval-Augmented Generation) capabilities, allowing the agent to access and use external knowledge sources.

Basic Knowledge Base Integration

from upsonic import KnowledgeBase
from upsonic.embeddings import OpenAIEmbeddingProvider
from upsonic.vectordb import ChromaVectorDB

# Create knowledge base
embedding_provider = OpenAIEmbeddingProvider()
vectordb = ChromaVectorDB()

knowledge_base = KnowledgeBase(
    sources=["document1.pdf", "document2.txt"],
    embedding_provider=embedding_provider,
    vectordb=vectordb
)

# Task with knowledge base context
task = Task(
    description="Answer questions about the uploaded documents",
    context=[knowledge_base]
)

Multiple Knowledge Bases

# Create multiple knowledge bases
kb1 = KnowledgeBase(
    sources=["technical_docs/"],
    embedding_provider=embedding_provider,
    vectordb=vectordb,
    name="Technical Documentation"
)

kb2 = KnowledgeBase(
    sources=["company_policies.pdf"],
    embedding_provider=embedding_provider,
    vectordb=vectordb,
    name="Company Policies"
)

# Task with multiple knowledge bases
task = Task(
    description="Find information about both technical procedures and company policies",
    context=[kb1, kb2]
)

Knowledge Base with Direct Content

# Knowledge base with direct string content
knowledge_base = KnowledgeBase(
    sources=["This is important information about our product features and capabilities."],
    embedding_provider=embedding_provider,
    vectordb=vectordb
)

task = Task(
    description="What are the key features mentioned in the product information?",
    context=[knowledge_base]
)

Adding Files and Images to Tasks

Tasks can include files, images, and documents through the context attribute. The framework automatically handles file loading and provides them to the model for analysis.

Single Image/File

# Task with single image
task = Task(
    description="Describe what you see in the attached image",
    context=["image.png"]
)

Multiple Files and Images

# Task with multiple files
task = Task(
    description="Compare the two attached images and analyze the document",
    context=["image1.jpg", "image2.jpg", "document.pdf"]
)

Files with Other Context

# Task with files and other context
task = Task(
    description="Analyze the attached chart and provide insights based on the market data",
    context=["chart.png", market_data_task, "additional_data.csv"]
)

Supported File Formats

The framework supports various file formats including:

Images: PNG (.png), JPEG (.jpg, .jpeg), GIF (.gif), BMP (.bmp), TIFF (.tiff)
Documents: PDF (.pdf), Word (.docx), Text (.txt), Markdown (.md)
Data: CSV (.csv), JSON (.json), XML (.xml), Excel (.xlsx)

# Example with different file formats
task = Task(
    description="Analyze all the attached files",
    context=["photo.jpg", "diagram.png", "data.csv", "report.pdf"]
)

Response Format

Tasks support various response formats to structure the output according to your needs. You can specify the expected format using the response_format parameter.

String Response (Default)

# Default string response
task = Task(
    description="Provide a simple text response",
    response_format=str
)

Pydantic Model Response

from pydantic import BaseModel

class AnalysisResult(BaseModel):
    summary: str
    confidence: float
    recommendations: list[str]
    key_metrics: dict[str, float]

# Task with structured response
task = Task(
    description="Analyze the provided data and provide structured results",
    response_format=AnalysisResult
)

Complex Nested Models

from pydantic import BaseModel
from typing import List, Optional

class Metric(BaseModel):
    name: str
    value: float
    unit: str

class Recommendation(BaseModel):
    title: str
    description: str
    priority: str
    estimated_impact: float

class DetailedAnalysis(BaseModel):
    summary: str
    confidence: float
    metrics: List[Metric]
    recommendations: List[Recommendation]
    risk_factors: Optional[List[str]] = None

# Task with complex structured response
task = Task(
    description="Perform comprehensive analysis with detailed metrics and recommendations",
    response_format=DetailedAnalysis
)

Accessing Task Results

After task execution, you can access various aspects of the task results and metadata through the task object properties and methods.

Basic Result Access

# Execute task
task = Task(description="What is the capital of France?")
result = agent.do(task)

# Access the response
print(f"Task result: {result}")
print(f"Task response: {task.response}")

Task Metadata

# Access task metadata
print(f"Task ID: {task.task_id}")
print(f"Price ID: {task.price_id}")
print(f"Duration: {task.duration}")
print(f"Start time: {task.start_time}")
print(f"End time: {task.end_time}")

Cost Information

# Access cost information
total_cost = task.total_cost
input_tokens = task.total_input_token
output_tokens = task.total_output_token

print(f"Total cost: ${total_cost}")
print(f"Input tokens: {input_tokens}")
print(f"Output tokens: {output_tokens}")

Tool Call History

# Access tool call history
tool_calls = task.tool_calls
for i, tool_call in enumerate(tool_calls):
    print(f"Tool call {i+1}:")
    print(f"  Tool: {tool_call.get('tool_name')}")
    print(f"  Parameters: {tool_call.get('params')}")
    print(f"  Result: {tool_call.get('tool_result')}")

Cache Information

# Access cache statistics
cache_stats = task.get_cache_stats()
print(f"Cache hit: {cache_stats.get('cache_hit')}")
print(f"Cache method: {cache_stats.get('cache_method')}")
print(f"Cache threshold: {cache_stats.get('cache_threshold')}")

Task State Information

# Check task state
print(f"Task is paused: {task.is_paused}")
print(f"Tools awaiting execution: {len(task.tools_awaiting_external_execution)}")

# For paused tasks, access external tool calls
if task.is_paused:
    for tool_call in task.tools_awaiting_external_execution:
        print(f"Tool: {tool_call.tool_name}")
        print(f"Arguments: {tool_call.tool_args}")
        print(f"Result: {tool_call.result}")

Complete Example

from upsonic import Agent, Task
from pydantic import BaseModel

class ReportResult(BaseModel):
    title: str
    summary: str
    key_points: list[str]
    confidence: float

# Create and execute task
agent = Agent(name="Analysis Agent")
task = Task(
    description="Generate a market analysis report",
    response_format=ReportResult,
    enable_cache=True
)

result = agent.do(task)

# Access all available information
print("=== TASK EXECUTION SUMMARY ===")
print(f"Task ID: {task.get_task_id()}")
print(f"Duration: {task.duration:.2f} seconds")
print(f"Cost: ${task.total_cost}")
print(f"Tokens: {task.total_input_token} in, {task.total_output_token} out")
print(f"Tool calls made: {len(task.tool_calls)}")
print(f"Cache hit: {task._cache_hit}")

print("\n=== TASK RESULT ===")
print(f"Result: {result}")
print(f"Response type: {type(task.response)}")

print("\n=== CACHE STATISTICS ===")
cache_stats = task.get_cache_stats()
for key, value in cache_stats.items():
    print(f"{key}: {value}")

This comprehensive documentation covers all aspects of working with Tasks in the Upsonic framework, from basic creation to advanced features like caching, knowledge base integration, and result access.

GET STARTED

Guides | 7 Step

CONCEPTS

DEPLOYMENT

FURTHER READINGS

​Overview

​Task Attributes

​Core Attributes

​Advanced Configuration

​Creating a Task

​Basic Task Creation

​Task with Tools

​Task with Response Format

​Adding Tools to a Task

​Single Tool

​Multiple Tools

​Tool Collections

​Tool Configuration

​Adding Tasks to Other Tasks in Context

​Basic Task Chaining

​Multiple Context Sources

​Complex Workflow Example

​Putting Knowledge Base to Task

​Basic Knowledge Base Integration

​Multiple Knowledge Bases

​Knowledge Base with Direct Content

​Adding Files and Images to Tasks

​Single Image/File

​Multiple Files and Images

​Files with Other Context

​Supported File Formats

​Response Format

​String Response (Default)

​Pydantic Model Response

​Complex Nested Models

​Accessing Task Results

​Basic Result Access

​Task Metadata

​Cost Information

​Tool Call History

​Cache Information

​Task State Information

​Complete Example