The complete, step-by-step roadmap to master AI Engineering.

Generative AI:  "Write me a poem about the ocean" → poem (done, no memory)
AI Agent:       "Find me flights under $500" → LLM calls a flight-search tool,
                 reads results, decides next action, loops until done
Agentic AI:     "Plan and book my entire trip" → multiple agents (research,
                 booking, budget-check) coordinate, remember context across
                 days, and adjust the plan if a flight price changes

def fetch_user_records(user_ids: list[int], *, active_only: bool = True) -> list[dict]:
    records = []
    for uid in user_ids:
        try:
            record = {"id": uid, "active": True}  # placeholder for real lookup
            if active_only and not record["active"]:
                continue
            records.append(record)
        except KeyError as e:
            raise ValueError(f"Malformed record for user {uid}") from e
    return records

# List comprehension instead of a manual loop
active_ids = [r["id"] for r in fetch_user_records([1, 2, 3]) if r["active"]]

from dataclasses import dataclass, field

@dataclass
class Tool:
    name: str
    description: str

@dataclass
class Agent:
    name: str
    tools: list[Tool] = field(default_factory=list)

    def add_tool(self, tool: Tool) -> None:
        self.tools.append(tool)

    def __repr__(self) -> str:
        return f"Agent(name={self.name!r}, tools={[t.name for t in self.tools]})"

search_tool = Tool(name="web_search", description="Searches the web")
agent = Agent(name="ResearchBot")
agent.add_tool(search_tool)
print(agent)  # Agent(name='ResearchBot', tools=['web_search'])

import httpx
from pydantic import BaseModel

class WeatherResponse(BaseModel):
    city: str
    temp_celsius: float

def get_weather(city: str) -> WeatherResponse:
    try:
        response = httpx.get(
            "https://api.example.com/weather",
            params={"city": city},
            headers={"Authorization": "Bearer YOUR_API_KEY"},
            timeout=5.0,
        )
        response.raise_for_status()
        return WeatherResponse(**response.json())
    except httpx.TimeoutException:
        raise RuntimeError("Weather API timed out")
    except httpx.HTTPStatusError as e:
        raise RuntimeError(f"Weather API failed: {e.response.status_code}")

import asyncio
import httpx

async def call_llm(client: httpx.AsyncClient, provider: str) -> dict:
    try:
        response = await asyncio.wait_for(
            client.post(f"https://api.{provider}.com/generate", json={"prompt": "hi"}),
            timeout=3.0,
        )
        return {"provider": provider, "result": response.json()}
    except asyncio.TimeoutError:
        return {"provider": provider, "result": None, "error": "timeout"}

async def main():
    async with httpx.AsyncClient() as client:
        results = await asyncio.gather(
            call_llm(client, "openai"),
            call_llm(client, "anthropic"),
            call_llm(client, "cohere"),
        )
        return results

# All three providers are called concurrently — slow one doesn't block the rest
asyncio.run(main())

from fastapi import FastAPI, HTTPException
from pydantic import BaseModel
import asyncio

app = FastAPI()

class PromptRequest(BaseModel):
    prompt: str

@app.post("/ask-all")
async def ask_all_providers(req: PromptRequest):
    try:
        results = await asyncio.wait_for(
            asyncio.gather(
                call_llm("openai", req.prompt),
                call_llm("anthropic", req.prompt),
            ),
            timeout=5.0,
        )
        return {"results": results}
    except asyncio.TimeoutError:
        raise HTTPException(status_code=504, detail="LLM providers timed out")

# Run with: uvicorn main:app --reload
# Auto docs available at /docs

from sqlmodel import SQLModel, Field, Session, create_engine, select

class AgentLog(SQLModel, table=True):
    id: int | None = Field(default=None, primary_key=True)
    agent_name: str
    prompt: str
    response: str

engine = create_engine("postgresql://user:pass@localhost/agentdb")
SQLModel.metadata.create_all(engine)

def save_log(agent_name: str, prompt: str, response: str):
    with Session(engine) as session:
        log = AgentLog(agent_name=agent_name, prompt=prompt, response=response)
        session.add(log)
        session.commit()

def get_logs_for_agent(agent_name: str) -> list[AgentLog]:
    with Session(engine) as session:
        return session.exec(
            select(AgentLog).where(AgentLog.agent_name == agent_name)
        ).all()