AI Agents Development

import asyncio
import httpx
from typing import List

# old way (synchronous) — slow
def fetch_sync(urls: List[str]):
    results = []
    with httpx.Client() as client:
        for url in urls:
            response = client.get(url)
            results.append(response.json())
    return results

# new way (asynchronous) — 10x faster
async def fetch_async(urls: List[str]):
    async with httpx.AsyncClient() as client:
        tasks = [client.get(url) for url in urls]
        responses = await asyncio.gather(*tasks)
        return [r.json() for r in responses]

# run:
# sync: takes ~5 seconds for 5 URLs
# async: takes ~1 second for 5 URLs

async def safe_fetch(url: str) -> dict | None:
    try:
        async with httpx.AsyncClient(timeout=10.0) as client:
            response = await client.get(url)
            response.raise_for_status()
            return response.json()
    except httpx.TimeoutException:
        print(f"Timeout: {url}")
        return None
    except httpx.HTTPStatusError as e:
        print(f"HTTP {e.response.status_code}: {url}")
        return None

# gather with return_exceptions=True — does not crash on a single error
results = await asyncio.gather(
    *[safe_fetch(url) for url in urls],
    return_exceptions=True
)
# filter valid results
valid = [r for r in results if r and not isinstance(r, Exception)]

import httpx
import asyncio
import json
from typing import AsyncGenerator

OPENAI_API_KEY = "sk-..."  # from the actual .env

# Streaming response from OpenAI
async def stream_openai(prompt: str) -> AsyncGenerator[str, None]:
    async with httpx.AsyncClient() as client:
        async with client.stream(
            "POST",
            "https://api.openai.com/v1/chat/completions",
            headers={"Authorization": f"Bearer {OPENAI_API_KEY}"},
            json={
                "model": "gpt-4o",
                "messages": [{"role": "user", "content": prompt}],
                "stream": True
            },
            timeout=60
        ) as response:
            async for line in response.aiter_lines():
                if line.startswith("data: ") and line != "data: [DONE]":
                    chunk = json.loads(line[6:])
                    if content := chunk["choices"][0]["delta"].get("content"):
                        yield content

# usage:
async def main():
    async for token in stream_openai("Tell me a short story"):
        print(token, end="", flush=True)

import asyncio
import httpx

async def robust_llm_call(payload: dict, max_retries: int = 3) -> dict:
    delay = 1.0
    for attempt in range(max_retries):
        try:
            async with httpx.AsyncClient(timeout=30) as client:
                resp = await client.post(
                    "https://api.openai.com/v1/chat/completions",
                    headers={"Authorization": f"Bearer {OPENAI_API_KEY}"},
                    json=payload
                )
                if resp.status_code == 429:  # Rate limited
                    retry_after = float(resp.headers.get("retry-after", delay))
                    await asyncio.sleep(retry_after)
                    delay *= 2
                    continue
                resp.raise_for_status()
                return resp.json()
        except httpx.ConnectError:
            if attempt == max_retries - 1:
                raise
            await asyncio.sleep(delay)
            delay *= 2
    raise RuntimeError("All attempts failed")

from pydantic import BaseModel, Field, validator
from typing import Optional, List
from enum import Enum
from datetime import datetime

class MessageRole(str, Enum):
    SYSTEM = "system"
    USER = "user"
    ASSISTANT = "assistant"

class Message(BaseModel):
    role: MessageRole
    content: str
    timestamp: datetime = Field(default_factory=datetime.now)

    @validator('content')
    def content_not_empty(cls, v):
        if not v.strip():
            raise ValueError('Content cannot be empty')
        return v.strip()

class ConversationConfig(BaseModel):
    model: str = "gpt-4o"
    temperature: float = Field(0.7, ge=0.0, le=2.0)
    max_tokens: int = Field(1024, gt=0, le=128000)
    system_prompt: Optional[str] = None
    memory_window: int = Field(10, ge=1, le=100)

class AgentResponse(BaseModel):
    content: str
    tool_calls: List[dict] = []
    tokens_used: int
    cost_usd: float
    duration_ms: int

from pydantic import BaseModel
from openai import AsyncOpenAI

class TaskBreakdown(BaseModel):
    title: str
    subtasks: list[str]
    estimated_hours: float
    complexity: str  # "low" | "medium" | "high"
    tools_needed: list[str]

client = AsyncOpenAI()

async def analyze_task(task_description: str) -> TaskBreakdown:
    response = await client.beta.chat.completions.parse(
        model="gpt-4o-mini",
        messages=[
            {"role": "system", "content": "Analyze tasks and break them into subtasks."},
            {"role": "user", "content": task_description}
        ],
        response_format=TaskBreakdown
    )
    return response.choices[0].message.parsed

# usage:
# result = await analyze_task("Build a chatbot with memory for a website")
# result.subtasks -> ["Define Pydantic models", "Build a FastAPI endpoint", ...]

from pydantic_settings import BaseSettings
from functools import lru_cache

class Settings(BaseSettings):
    # AI APIs
    openai_api_key: str
    anthropic_api_key: str = ""

    # Database
    database_url: str = "sqlite:///./app.db"
    redis_url: str = "redis://localhost:6379"

    # App
    debug: bool = False
    secret_key: str
    allowed_origins: list[str] = ["http://localhost:3000"]

    class Config:
        env_file = ".env"
        env_file_encoding = "utf-8"

@lru_cache()  # Singleton pattern — creates once
def get_settings() -> Settings:
    return Settings()

# sample .env file:
# OPENAI_API_KEY=sk-...
# SECRET_KEY=your-secret-key-here
# DEBUG=false

from langchain_openai import ChatOpenAI
from langchain_core.prompts import ChatPromptTemplate
from langchain_core.output_parsers import StrOutputParser, JsonOutputParser
from langchain_core.runnables import RunnablePassthrough, RunnableLambda

llm = ChatOpenAI(model="gpt-4o-mini", temperature=0)

# a simple Chain — LCEL syntax
prompt = ChatPromptTemplate.from_messages([
    ("system", "You are a professional AI assistant that answers in English."),
    ("human", "{question}")
])

chain = prompt | llm | StrOutputParser()

# run:
result = chain.invoke({"question": "What is the difference between AI and ML?"})

# a complex Chain with parallel branches
from langchain_core.runnables import RunnableParallel

map_chain = RunnableParallel(
    summary=prompt | llm | StrOutputParser(),
    sentiment=ChatPromptTemplate.from_template(
        "Rate the sentiment of: {question}. Return JSON: {{score: 1-10}}"
    ) | llm | JsonOutputParser()
)

result = map_chain.invoke({"question": "Your product is great!"})
# → {"summary": "...", "sentiment": {"score": 9}}

async def stream_response(question: str):
    async for chunk in chain.astream({"question": question}):
        print(chunk, end="", flush=True)

# Batch processing — hundreds of questions in parallel
questions = [{"question": q} for q in ["question 1", "question 2", "question 3"]]
results = await chain.abatch(questions, config={"max_concurrency": 5})

from langchain.memory import (
    ConversationBufferMemory,
    ConversationSummaryMemory,
    ConversationBufferWindowMemory,
    ConversationEntityMemory
)

# Buffer Window — remembers the last N messages
memory = ConversationBufferWindowMemory(
    k=5,  # remembers 5 question-answer pairs
    return_messages=True,
    memory_key="chat_history"
)

# Summary Memory — summarizes long conversations
summary_memory = ConversationSummaryMemory(
    llm=ChatOpenAI(model="gpt-4o-mini"),
    return_messages=True
)

# Entity Memory — remembers facts about people and places
entity_memory = ConversationEntityMemory(
    llm=ChatOpenAI(model="gpt-4o-mini")
)
# "the user mentioned they live in Tel Aviv and work at a startup"

from langchain_community.document_loaders import PyPDFLoader
from langchain.text_splitter import RecursiveCharacterTextSplitter
from langchain_openai import OpenAIEmbeddings
from langchain_community.vectorstores import Chroma
from langchain.chains import RetrievalQA

# step 1: loading the PDF
loader = PyPDFLoader("company_docs.pdf")
documents = loader.load()

# step 2: splitting into chunks (Chunking)
splitter = RecursiveCharacterTextSplitter(
    chunk_size=1000,
    chunk_overlap=200,
    separators=["\n\n", "\n", ".", " "]
)
chunks = splitter.split_documents(documents)

# step 3: creating the Vector Store
embeddings = OpenAIEmbeddings()
vectorstore = Chroma.from_documents(
    chunks,
    embeddings,
    persist_directory="./chroma_db"
)

# step 4: creating the RAG Chain
qa_chain = RetrievalQA.from_chain_type(
    llm=ChatOpenAI(model="gpt-4o"),
    retriever=vectorstore.as_retriever(search_kwargs={"k": 4}),
    return_source_documents=True
)

result = qa_chain.invoke({"query": "What is our return policy?"})
print(result["result"])

from langchain_community.document_loaders import (
    PyPDFLoader,
    WebBaseLoader,
    JSONLoader,
    CSVLoader,
    UnstructuredWordDocumentLoader
)
from langchain_core.documents import Document

# loading multiple documents with metadata
async def load_documents(file_paths: list[str]) -> list[Document]:
    all_docs = []

    for path in file_paths:
        if path.endswith('.pdf'):
            loader = PyPDFLoader(path)
        elif path.endswith('.docx'):
            loader = UnstructuredWordDocumentLoader(path)
        elif path.endswith('.csv'):
            loader = CSVLoader(path)
        else:
            continue

        docs = loader.load()

        # add metadata to each document
        for doc in docs:
            doc.metadata['source_file'] = path
            doc.metadata['loaded_at'] = datetime.now().isoformat()

        all_docs.extend(docs)

    return all_docs