Learn AFK in 15 Minutes

from afk.agents import Agent
from afk.core import Runner

agent = Agent(
    name="tutor",
    model="gpt-5.2-mini",
    instructions="""
    You are a programming tutor.
    Explain concepts using simple language and short code examples.
    Keep answers under 100 words.
    """,
)

runner = Runner()
result = runner.run_sync(agent, user_message="What is recursion?")

print(result.final_text)   # The model's explanation
print(result.state)        # "completed"
print(result.run_id)       # Unique run ID for tracing

from pydantic import BaseModel
from afk.agents import Agent
from afk.tools import tool
from afk.core import Runner

class LookupArgs(BaseModel):
    topic: str
    max_results: int = 3

@tool(args_model=LookupArgs, name="search_docs", description="Search the documentation by topic.")
def search_docs(args: LookupArgs) -> dict:
    # In production, this would call a real search API
    return {
        "results": [
            {"title": f"Guide: {args.topic}", "relevance": 0.95},
            {"title": f"FAQ: {args.topic}", "relevance": 0.82},
        ][:args.max_results]
    }

agent = Agent(
    name="research-tutor",
    model="gpt-5.2-mini",
    instructions="Use search_docs to find relevant documentation before answering.",
    tools=[search_docs],   # ← Attach tools here
)

runner = Runner()
result = runner.run_sync(agent, user_message="How do I handle errors in Python?")

print(result.final_text)
print(f"Tool calls made: {len(result.tool_executions)}")

# Inspect what tools were called
for rec in result.tool_executions:
    print(f"  {rec.tool_name}: {'[OK]' if rec.success else '[ERR]'} ({rec.latency_ms:.0f}ms)")

import asyncio
from afk.agents import Agent
from afk.core import Runner

agent = Agent(
    name="explainer",
    model="gpt-5.2-mini",
    instructions="Give detailed, well-structured explanations.",
    tools=[search_docs],   # Reuse the tool from above
)

async def main():
    runner = Runner()
    handle = await runner.run_stream(
        agent, user_message="Explain the Python GIL and its impact on threading"
    )

    async for event in handle:
        match event.type:
            case "text_delta":
                print(event.text_delta, end="", flush=True)
            case "tool_started":
                print(f"\n[TOOL] Calling {event.tool_name}...")
            case "tool_completed":
                status = "[OK]" if event.success else "[ERR]"
                print(f"   {status} {event.tool_name} done")
            case "completed":
                print(f"\n\n[DONE] Run completed ({event.result.state})")

    # Access the full result after streaming
    result = handle.result
    print(f"Total tokens: {result.usage.total_tokens}")

asyncio.run(main())

import asyncio
from afk.agents import Agent
from afk.core import Runner

# Specialist 1: Research agent
researcher = Agent(
    name="researcher",
    model="gpt-5.2-mini",
    instructions="""
    You are a research specialist.
    Find relevant facts and present them as bullet points.
    Be thorough but concise.
    """,
)

# Specialist 2: Writing agent
writer = Agent(
    name="writer",
    model="gpt-5.2-mini",
    instructions="""
    You are a technical writer.
    Take research findings and write a clear, well-structured summary.
    Use simple language suitable for a general audience.
    """,
)

# Coordinator: delegates and combines
coordinator = Agent(
    name="coordinator",
    model="gpt-5.2-mini",
    instructions="""
    You manage a team:
    - Delegate research tasks to 'researcher'
    - Delegate writing tasks to 'writer'
    Combine their work into a polished final response.
    """,
    subagents=[researcher, writer],
)

async def main():
    runner = Runner()
    result = await runner.run(
        coordinator,
        user_message="Write a brief explainer about quantum computing for beginners",
    )

    print(result.final_text)
    print(f"\nSubagent calls: {len(result.subagent_executions)}")
    for sub in result.subagent_executions:
        print(f"  -> {sub.subagent_name}: {'[OK]' if sub.success else '[ERR]'}")

asyncio.run(main())

import asyncio
from afk.agents import Agent
from afk.core import Runner

agent = Agent(
    name="tutor",
    model="gpt-5.2-mini",
    instructions="You are a Python tutor. Remember what the student has asked before.",
)

async def main():
    runner = Runner()
    thread = "session-42"   # ← Same thread_id = same conversation

    # Turn 1
    r1 = await runner.run(agent, user_message="What are decorators?", thread_id=thread)
    print(f"Turn 1: {r1.final_text[:80]}...")

    # Turn 2 — the agent remembers Turn 1
    r2 = await runner.run(agent, user_message="Give me an example", thread_id=thread)
    print(f"Turn 2: {r2.final_text[:80]}...")

    # Turn 3 — still has context
    r3 = await runner.run(agent, user_message="How is that different from a wrapper function?", thread_id=thread)
    print(f"Turn 3: {r3.final_text[:80]}...")

asyncio.run(main())