Attributes

from upsonic import Agent, Task
from upsonic.tools import tool
from pydantic import BaseModel
from typing import List

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

# Create a simple function tool
@tool
def get_market_data(year: int, quarter: int) -> str:
    """Retrieve market data for a specific year and quarter."""
    # In a real scenario, this would fetch actual data
    return f"Market data for Q{quarter} {year}: Growth rate 5.2%, Market cap $2.5T"

agent = Agent("anthropic/claude-sonnet-4-5")

task = Task(
    description="Analyze the market trends for Q4 2024",
    response_format=AnalysisResult,
    enable_thinking_tool=True,
    tools=[get_market_data]  # Add the function tool to the task
)

result = agent.print_do(task)
print(result.summary)
print(f"Confidence: {result.confidence}")
print(f"Recommendations: {result.recommendations}")

from upsonic import Agent, Task
from upsonic.embeddings.factory import auto_detect_best_embedding

agent = Agent("anthropic/claude-sonnet-4-5")

# Get embedding provider for vector search caching
embedding_provider = auto_detect_best_embedding()

task = Task(
    description="Analyze the market trends for Q4 2024",
    enable_cache=True,
    cache_method="vector_search",
    cache_threshold=0.8,
    cache_duration_minutes=60,
    cache_embedding_provider=embedding_provider
)

agent.print_do(task)

# Check if response came from cache
if task.cache_hit:
    print("Response retrieved from cache!")

# Get cache statistics
stats = task.get_cache_stats()
print(f"Cache hit rate: {stats['hit_rate']:.2%}")

from upsonic import Agent, Task

def validate_output(output: str) -> bool:
    """Validate that output contains required information."""
    required_keywords = ["summary", "analysis", "conclusion"]
    return all(keyword in output.lower() for keyword in required_keywords)

agent = Agent("anthropic/claude-sonnet-4-5")

task = Task(
    description="Write a comprehensive market analysis report",
    guardrail=validate_output,
    guardrail_retries=3
)

agent.print_do(task)

from upsonic import Agent, Task

agent = Agent("anthropic/claude-sonnet-4-5")

# Files in context are automatically extracted to attachments
task = Task(
    description="Summarize the key points from these documents",
    context=["/path/to/document1.pdf", "/path/to/document2.pdf", "/path/to/image.png"]
)

agent.print_do(task)

from upsonic import Agent, Task

agent = Agent("anthropic/claude-sonnet-4-5")

task = Task(
    description="Search for relevant information about AI agents",
    vector_search_top_k=10,
    vector_search_alpha=0.7,
    vector_search_fusion_method="rrf",
    vector_search_similarity_threshold=0.6,
    vector_search_filter={"category": "technology", "year": 2024}
)

agent.print_do(task)

from upsonic import Agent, Task

agent = Agent("anthropic/claude-sonnet-4-5")

task = Task(
    description="Perform a complex analysis"
)

agent.print_do(task)

# Access task properties
print(f"Task ID:      {task.id}")
if task.start_time and task.end_time:
    print(f"Wall-clock:   {(task.end_time - task.start_time):.2f} seconds")
print(f"Status:       {task.status}")
u = task.usage
if u.cost is not None:
    print(f"Total Cost:   ${u.cost:.4f}")
print(f"Input Tokens: {u.input_tokens}")
print(f"Output Tokens:{u.output_tokens}")
print(f"Tool Calls:   {len(task.tool_calls)}")

# Check task state
if task.is_completed:
    print("Task completed successfully")
elif task.is_problematic:
    print("Task has issues and needs continuation")