> ## Documentation Index
> Fetch the complete documentation index at: https://docs.upsonic.ai/llms.txt
> Use this file to discover all available pages before exploring further.

# Task-Centric Structure

> Detailed guide on managing and creating tasks within the Upsonic framework.

## 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

```python theme={null}
from upsonic import Task

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

### Task with Tools

```python theme={null}
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

```python theme={null}
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

```python theme={null}
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

```python theme={null}
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

```python theme={null}
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

```python theme={null}
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

```python theme={null}
# 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

```python theme={null}
# 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

```python theme={null}
# 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

```python theme={null}
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

```python theme={null}
# 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

```python theme={null}
# 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

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

### Multiple Files and Images

```python theme={null}
# 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

```python theme={null}
# 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)

```python theme={null}
# 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)

```python theme={null}
# Default string response
task = Task(
    description="Provide a simple text response",
    response_format=str
)
```

### Pydantic Model Response

```python theme={null}
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

```python theme={null}
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

```python theme={null}
# 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

```python theme={null}
# Access task metadata
print(f"Task ID:       {task.task_id}")
print(f"Task Usage ID: {task.task_usage_id}")
print(f"Start time:    {task.start_time}")
print(f"End time:      {task.end_time}")
if task.start_time and task.end_time:
    print(f"Wall-clock:    {(task.end_time - task.start_time):.2f}s")
```

### Cost Information

```python theme={null}
# Access cost information via the unified usage view
u = task.usage
if u.cost is not None:
    print(f"Total cost:   ${u.cost:.6f}")
print(f"Input tokens:  {u.input_tokens}")
print(f"Output tokens: {u.output_tokens}")
print(f"Requests:      {u.requests}")
```

### Tool Call History

```python theme={null}
# 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

```python theme={null}
# 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

```python theme={null}
# 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

```python theme={null}
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()}")
if task.start_time and task.end_time:
    print(f"Wall-clock:     {(task.end_time - task.start_time):.2f} seconds")
u = task.usage
if u.cost is not None:
    print(f"Cost:           ${u.cost:.6f}")
print(f"Tokens:         {u.input_tokens} in, {u.output_tokens} out")
print(f"Tool calls:     {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.