Agents

from agnt5 import Agent, LanguageModel

lm = LanguageModel()

agent = Agent(
    name="assistant",
    model=lm,
    instructions="You are a helpful coding assistant."
)

# Run agent
result = await agent.run("Explain recursion")
print(result.output)

from agnt5 import Agent, tool, LanguageModel

@tool.function(auto_schema=True)
def search_docs(query: str, language: str = "python") -> List[Dict]:
    """Search programming language documentation."""
    # Implementation
    return search_results

@tool.function(auto_schema=True)
def run_code(code: str, language: str = "python") -> Dict[str, str]:
    """Execute code and return output."""
    # Implementation
    return {"output": result}

lm = LanguageModel()
agent = Agent(
    name="coding_assistant",
    model=lm,
    instructions="""You are a coding assistant.
    Use search_docs to find API references.
    Use run_code to test code examples.""",
    tools=[search_docs, run_code]
)

result = await agent.run("How do I read a file in Python? Show me an example.")

from agnt5 import Agent, Session, Memory, LanguageModel

# Create session
session = Session(
    id="tutoring-session-789",
    user_id="student-123",
    metadata={"subject": "mathematics"}
)

# Create memory
memory = Memory(service=VectorMemoryService())
await memory.store("student_level", "Advanced calculus")

# Create agent
lm = LanguageModel()
agent = Agent(
    name="math_tutor",
    model=lm,
    instructions="You are a patient math tutor. Adapt to student's level.",
    tools=[solve_equation_tool, plot_function_tool],
    session=session,
    memory=memory
)

result = await agent.run("Help me understand limits")

async for event in agent.stream("Analyze this dataset", session=session):
    match event.type:
        case "thinking":
            print(f"🤔 {event.content}")
        case "tool_call":
            print(f"🔧 Calling {event.tool_name}({event.arguments})")
        case "tool_result":
            print(f"✓ Result: {event.result}")
        case "response":
            print(f"💬 {event.content}")
        case "error":
            print(f"❌ Error: {event.error}")

# Get plan without executing
plan = agent.plan("Analyze competitor pricing strategies")

print(f"Estimated steps: {len(plan.steps)}")
for step in plan.steps:
    print(f"- {step.type}: {step.description}")
    if step.tool:
        print(f"  Tool: {step.tool.name}")

# Review and execute if approved
if user_approves(plan):
    result = await agent.run("Analyze competitor pricing strategies")

from agnt5 import Agent, Session, Memory, tool

@tool.function(auto_schema=True)
def search_academic(query: str, year_from: int = 2020) -> List[Dict]:
    """Search academic papers."""
    pass

@tool.function(auto_schema=True)
def extract_insights(paper_text: str) -> Dict[str, List[str]]:
    """Extract key insights from paper."""
    pass

# Create research agent
session = Session(id="research-ai-safety-001", user_id="researcher-123")
memory = Memory(service=VectorMemoryService())

lm = LanguageModel()
research_agent = Agent(
    name="research_agent",
    model=lm,
    instructions="""You are a research assistant specializing in AI safety.

    Research process:
    1. Search for relevant recent papers
    2. Extract key insights from each paper
    3. Identify common themes and gaps
    4. Synthesize findings into comprehensive summary""",
    tools=[search_academic, extract_insights],
    session=session,
    memory=memory
)

result = await research_agent.run(
    "Survey the current state of AI alignment research"
)

# Store findings in memory
await memory.ingest_from_session(session, strategy="smart")

# Shared session for coordination
session = Session(id="product-launch-001", user_id="pm-456")

# Specialized agents
market_researcher = Agent(
    name="market_analyst",
    model=lm,
    tools=[market_data_tool, competitor_analysis_tool],
    session=session,
    instructions="Analyze market opportunities and competitive landscape."
)

product_designer = Agent(
    name="designer",
    model=lm,
    tools=[design_tool, user_research_tool],
    session=session,
    instructions="Design products based on market research and user needs."
)

technical_lead = Agent(
    name="tech_lead",
    model=lm,
    tools=[architecture_tool, feasibility_tool],
    session=session,
    instructions="Assess technical feasibility and propose architecture."
)

# Sequential execution with shared context
market_analysis = await market_researcher.run(
    "Analyze market for AI-powered code review tools"
)

product_specs = await product_designer.run(
    "Design product based on market analysis"
)

tech_assessment = await technical_lead.run(
    "Evaluate technical feasibility of proposed product"
)

from agnt5.tools import AgentTool

# Create specialized agents
billing_agent = Agent(
    name="billing_specialist",
    model=lm,
    tools=[payment_tool, invoice_tool, refund_tool],
    instructions="Handle billing, payments, and refunds."
)

technical_agent = Agent(
    name="tech_support",
    model=lm,
    tools=[diagnostic_tool, fix_tool],
    instructions="Diagnose and fix technical issues."
)

# Coordinator with handoff capability
coordinator = Agent(
    name="coordinator",
    model=lm,
    tools=[
        classify_request_tool,
        AgentTool(target_agent=billing_agent),
        AgentTool(target_agent=technical_agent)
    ],
    instructions="""You are a support coordinator.
    Classify requests and hand off to appropriate specialist.

    Hand off to:
    - billing_specialist: payment, invoice, refund questions
    - tech_support: technical issues, bugs, troubleshooting"""
)

session = Session(id="support-ticket-789", user_id="customer-123")
result = await coordinator.run(
    "I was charged twice for my subscription",
    session=session
)

@tool.function(auto_schema=True, confirmation=True)
def deploy_to_production(version: str) -> Dict[str, str]:
    """Deploy application to production.

    Warning: Requires human approval.
    """
    pass

deployment_agent = Agent(
    name="deployer",
    model=lm,
    tools=[run_tests_tool, deploy_to_production],
    instructions="""Run all tests before deploying.
    Always request human approval for production deployments."""
)

result = await deployment_agent.run("Deploy version 2.0 to production")
# Agent runs tests, then waits for human approval before deploying

debugging_agent = Agent(
    name="debugger",
    model=lm,
    tools=[
        analyze_logs_tool,
        run_diagnostic_tool,
        apply_fix_tool,
        verify_fix_tool
    ],
    instructions="""You are a debugging assistant.

    Process:
    1. Analyze error logs to identify root cause
    2. Run diagnostics to confirm hypothesis
    3. Apply potential fix
    4. Verify fix works
    5. If not fixed, iterate (max 3 attempts)

    Always verify fixes before considering issue resolved."""
)

result = await debugging_agent.run(
    "Users are experiencing 500 errors on the checkout page"
)

# ✓ Good - Specific process
agent = Agent(
    name="analyst",
    instructions="""Analyze data systematically:
    1. Identify data patterns and anomalies
    2. Calculate key statistics
    3. Generate visualizations
    4. Provide actionable insights"""
)

# ✗ Bad - Too vague
agent = Agent(
    name="helper",
    instructions="Help with analysis"
)

# Create shared session
session = Session(id="project-workflow-123", user_id="user-456")

# Set shared context
session.set_state("project_name", "ai-safety-research")
session.set_state("deadline", "2024-12-31")

# All agents access shared context
agent1 = Agent(name="agent1", session=session, ...)
agent2 = Agent(name="agent2", session=session, ...)

# Store user preferences
await memory.store("user_expertise", "Expert in React and TypeScript")
await memory.store("coding_style", "Prefers functional programming")

# Agent recalls automatically
agent = Agent(
    name="assistant",
    model=lm,
    tools=[code_gen_tool],
    memory=memory
)

result = await agent.run("Help me build a component")
# Agent uses stored preferences

agent = Agent(
    name="bounded_agent",
    model=lm,
    max_iterations=5,  # Stop after 5 reasoning loops
    instructions="Solve problems efficiently."
)

Agent
├── LanguageModel (reasoning)
├── Tools (actions via Function)
├── Session (context via Entity)
├── Memory (knowledge via Entity)
└── Planner (orchestration)