Agentic AI

from openai import OpenAI
import json

client = OpenAI()

# Define tools as JSON schema
tools = [
    {
        "type": "function",
        "function": {
            "name": "search_web",
            "description": "Search the web for current information. Use when you need facts you don't know.",
            "parameters": {
                "type": "object",
                "properties": {
                    "query": {
                        "type": "string",
                        "description": "The search query"
                    },
                    "num_results": {
                        "type": "integer",
                        "description": "Number of results to return (1-10)",
                        "default": 5
                    }
                },
                "required": ["query"]
            }
        }
    },
    {
        "type": "function",
        "function": {
            "name": "run_python",
            "description": "Execute Python code and return the output. Use for calculations, data processing.",
            "parameters": {
                "type": "object",
                "properties": {
                    "code": {"type": "string", "description": "Python code to execute"}
                },
                "required": ["code"]
            }
        }
    }
]

import json
from openai import OpenAI

client = OpenAI()
MAX_ITERATIONS = 10  # Safety limit

def run_agent(task: str) -> str:
    messages = [
        {"role": "system", "content": "You are a helpful research assistant. Use tools to answer questions accurately."},
        {"role": "user", "content": task}
    ]

    for iteration in range(MAX_ITERATIONS):
        response = client.chat.completions.create(
            model="gpt-4o",
            messages=messages,
            tools=tools,
            tool_choice="auto",   # Let model decide
        )

        msg = response.choices[0].message
        messages.append(msg)  # Add assistant turn to history

        # No tool calls → agent is done
        if not msg.tool_calls:
            return msg.content

        # Execute each tool call
        for call in msg.tool_calls:
            fn_name = call.function.name
            fn_args = json.loads(call.function.arguments)

            print(f"[Agent] Calling {fn_name}({fn_args})")
            result = dispatch_tool(fn_name, fn_args)  # Your tool executor

            messages.append({
                "role": "tool",
                "content": json.dumps(result),
                "tool_call_id": call.id
            })

    return "Max iterations reached without completing the task."

# Run it
answer = run_agent("What are the top 3 Python web frameworks in 2025?")
print(answer)

import subprocess
import requests

def dispatch_tool(name: str, args: dict) -> dict:
    """Route tool calls to actual implementations with error handling."""
    try:
        if name == "search_web":
            return search_web(**args)
        elif name == "run_python":
            return run_python(**args)
        elif name == "read_file":
            return read_file(**args)
        else:
            return {"error": f"Unknown tool: {name}"}
    except Exception as e:
        return {"error": str(e), "tool": name}

def run_python(code: str) -> dict:
    """Execute Python in a subprocess sandbox."""
    try:
        result = subprocess.run(
            ["python3", "-c", code],
            capture_output=True, text=True,
            timeout=10  # Never run untrusted code without a timeout!
        )
        return {
            "stdout": result.stdout,
            "stderr": result.stderr,
            "exit_code": result.returncode
        }
    except subprocess.TimeoutExpired:
        return {"error": "Code execution timed out (10s limit)"}

def read_file(path: str) -> dict:
    """Read file with path validation."""
    from pathlib import Path
    safe_root = Path("/workspace")  # Restrict to safe directory
    target = (safe_root / path).resolve()
    if not str(target).startswith(str(safe_root)):
        return {"error": "Path traversal blocked"}
    return {"content": target.read_text()}

from openai import OpenAI
import psycopg2
import json

client = OpenAI()

def embed(text: str) -> list[float]:
    """Create embedding for text."""
    res = client.embeddings.create(model="text-embedding-3-small", input=text)
    return res.data[0].embedding

def save_memory(conn, user_id: str, content: str, memory_type: str = "episodic"):
    """Save a memory to the vector store."""
    vector = embed(content)
    with conn.cursor() as cur:
        cur.execute(
            "INSERT INTO agent_memory (user_id, content, embedding, type) VALUES (%s, %s, %s, %s)",
            (user_id, content, json.dumps(vector), memory_type)
        )
    conn.commit()

def recall_memory(conn, user_id: str, query: str, top_k: int = 5) -> list[str]:
    """Retrieve relevant memories using cosine similarity."""
    vector = embed(query)
    with conn.cursor() as cur:
        cur.execute(
            """
            SELECT content
            FROM agent_memory
            WHERE user_id = %s
            ORDER BY embedding <=> %s::vector
            LIMIT %s
            """,
            (user_id, json.dumps(vector), top_k)
        )
        return [row[0] for row in cur.fetchall()]

# Use in agent
def build_system_prompt(user_id: str, task: str, conn) -> str:
    memories = recall_memory(conn, user_id, task)
    memory_block = "\n".join(f"- {m}" for m in memories)
    return f"""You are a personal assistant.

Relevant memories about this user:
{memory_block}

Use these to personalize your response."""

# ReAct forces the model to reason before each action
# Paper: https://arxiv.org/abs/2210.03629

REACT_SYSTEM_PROMPT = """
You are a research agent. Solve tasks step by step.

For each step, use this EXACT format:

Thought: [reason about what to do next]
Action: [tool_name]
Action Input: {"param": "value"}
Observation: [tool result will be inserted here]

... repeat Thought/Action/Observation as needed ...

Final Answer: [your complete answer to the original question]

IMPORTANT:
- Only call ONE tool per step
- Always check the Observation before the next Thought
- Stop when you have enough information for a Final Answer
"""

# Example trace:
# Thought: I need to find current Python popularity stats
# Action: search_web
# Action Input: {"query": "Python most popular language 2025"}
# Observation: {"results": [{"title": "Python #1 for 4th year...", ...}]}
# Thought: I have the data I need
# Final Answer: Python remains the #1 language in 2025...

from openai import OpenAI
import json

client = OpenAI()

def plan(task: str) -> list[str]:
    """Step 1: Generate a plan as a list of steps."""
    response = client.chat.completions.create(
        model="gpt-4o",
        messages=[
            {"role": "system", "content": "You are a planning assistant. Break complex tasks into ordered, concrete steps. Return JSON: {steps: [str]}."},
            {"role": "user", "content": f"Task: {task}"}
        ],
        response_format={"type": "json_object"}
    )
    return json.loads(response.choices[0].message.content)["steps"]

def execute_step(step: str, context: str) -> str:
    """Step 2: Execute one step with tools."""
    # Run single-step agent with ReAct
    return run_agent(f"Context so far:\n{context}\n\nCurrent step: {step}")

def replan(original_task: str, steps: list, completed: list, last_result: str) -> list[str]:
    """Step 3: Replan if execution reveals new information."""
    response = client.chat.completions.create(
        model="gpt-4o",
        messages=[{
            "role": "user",
            "content": f"Task: {original_task}\nCompleted: {completed}\nRemaining: {steps}\nLast result: {last_result}\n\nDo remaining steps still make sense? Return updated plan as JSON: {{steps: [str]}}"
        }],
        response_format={"type": "json_object"}
    )
    return json.loads(response.choices[0].message.content)["steps"]

# Run the full pattern
task = "Research and write a comparison of the top 3 vector databases"
steps = plan(task)
results = []
for step in steps:
    result = execute_step(step, "\n".join(results))
    results.append(result)
    steps = replan(task, steps[steps.index(step)+1:], steps[:steps.index(step)+1], result)

def reflect_and_improve(task: str, initial_output: str, criteria: list[str]) -> str:
    """
    Reflexion pattern: agent critiques its own output
    and produces an improved version.
    Paper: https://arxiv.org/abs/2303.11366
    """
    criteria_str = "\n".join(f"- {c}" for c in criteria)

    # Step 1: Self-critique
    critique = client.chat.completions.create(
        model="gpt-4o",
        messages=[{
            "role": "user",
            "content": f"""Task: {task}

Output to review:
{initial_output}

Evaluate against these criteria:
{criteria_str}

List specific issues and improvements needed. Be critical."""
        }]
    ).choices[0].message.content

    # Step 2: Revise based on critique
    improved = client.chat.completions.create(
        model="gpt-4o",
        messages=[{
            "role": "user",
            "content": f"""Original task: {task}
First attempt: {initial_output}
Critique: {critique}

Now produce an improved version that addresses all issues."""
        }]
    ).choices[0].message.content

    return improved

# Usage
result = reflect_and_improve(
    task="Write a Python function to parse CSV files",
    initial_output=agent_output,
    criteria=["Handles edge cases", "Has error handling", "Includes type hints", "Has docstring"]
)

from langgraph.graph import StateGraph, END
from langchain_openai import ChatOpenAI
from typing import TypedDict, Literal
import operator
from typing import Annotated

llm = ChatOpenAI(model="gpt-4o")

class State(TypedDict):
    task: str
    plan: list[str]
    results: Annotated[list[str], operator.add]  # Accumulate results
    final_output: str
    next: str

# --- Specialized agents ---
def researcher(state: State) -> State:
    """Searches and collects information."""
    result = run_agent_with_tools(state["task"], tools=[search_web, scrape_url])
    return {"results": [f"Research: {result}"], "next": "writer"}

def writer(state: State) -> State:
    """Synthesizes research into a draft."""
    context = "\n".join(state["results"])
    result = llm.invoke(f"Write a report based on:\n{context}\n\nTask: {state['task']}")
    return {"results": [f"Draft: {result.content}"], "next": "reviewer"}

def reviewer(state: State) -> State:
    """Reviews and improves the draft."""
    draft = state["results"][-1]
    reviewed = llm.invoke(f"Review and improve this draft:\n{draft}")
    return {"final_output": reviewed.content, "next": END}

def supervisor(state: State) -> Literal["researcher", "writer", "reviewer", "__end__"]:
    """Routes to next agent based on state."""
    return state.get("next", "researcher")

# --- Build graph ---
graph = StateGraph(State)
graph.add_node("researcher", researcher)
graph.add_node("writer", writer)
graph.add_node("reviewer", reviewer)
graph.set_entry_point("researcher")
graph.add_conditional_edges("researcher", supervisor)
graph.add_conditional_edges("writer", supervisor)
graph.add_conditional_edges("reviewer", supervisor)

app = graph.compile()
result = app.invoke({"task": "Write a report on the state of AI agents in 2025"})
print(result["final_output"])

# Option 1: E2B — cloud sandboxes (recommended for production)
# https://e2b.dev
from e2b_code_interpreter import Sandbox

with Sandbox() as sbx:
    execution = sbx.run_code("""
import pandas as pd
df = pd.DataFrame({"a": [1,2,3], "b": [4,5,6]})
print(df.describe())
""")
    print(execution.text)   # stdout
    print(execution.error)  # stderr

# Option 2: Docker subprocess (self-hosted)
import docker

client = docker.from_env()

def run_in_docker(code: str) -> str:
    result = client.containers.run(
        image="python:3.12-slim",
        command=["python", "-c", code],
        mem_limit="128m",
        cpu_period=100000,
        cpu_quota=50000,    # 50% CPU
        network_disabled=True,  # No network access!
        remove=True,
        timeout=10
    )
    return result.decode()

# Option 3: Pyodide (browser/WASM — zero server risk)
# https://pyodide.org
# Run Python in browser — no server execution

# Tavily — best for agents (structured, fast, cheap)
# https://tavily.com
from tavily import TavilyClient
client = TavilyClient(api_key="tvly-...")
result = client.search("LangGraph tutorial 2025", max_results=5)
# Returns: {results: [{title, url, content, score}]}

# Brave Search API (privacy-focused alternative)
# https://api.search.brave.com
import requests
res = requests.get(
    "https://api.search.brave.com/res/v1/web/search",
    headers={"X-Subscription-Token": "your-key"},
    params={"q": "your query", "count": 5}
)

# Serper (Google results via API)
# https://serper.dev
import requests
res = requests.post(
    "https://google.serper.dev/search",
    headers={"X-API-KEY": "your-key"},
    json={"q": "your query"}
)

# Browser automation — Playwright (for JS-heavy sites)
# https://playwright.dev
from playwright.async_api import async_playwright
async with async_playwright() as p:
    browser = await p.chromium.launch()
    page = await browser.new_page()
    await page.goto("https://example.com")
    content = await page.content()

# File operations tool
def read_file(path: str, start_line: int = 0, end_line: int = 100) -> dict:
    """Read file with line range to avoid dumping large files into context."""
    lines = open(path).readlines()
    return {
        "content": "".join(lines[start_line:end_line]),
        "total_lines": len(lines),
        "showing": f"{start_line}-{end_line}"
    }

def write_file(path: str, content: str) -> dict:
    with open(path, "w") as f:
        f.write(content)
    return {"success": True, "path": path}

# SQL database tool
import psycopg2
def query_db(sql: str, params: list = None) -> dict:
    """Run read-only SQL queries."""
    # IMPORTANT: Use read-only DB user in production!
    conn = psycopg2.connect(DATABASE_URL)
    with conn.cursor() as cur:
        cur.execute(sql, params or [])
        rows = cur.fetchall()
        cols = [d[0] for d in cur.description]
    return {"columns": cols, "rows": rows[:50]}  # Limit rows

# HTTP / REST API tool
import httpx
async def call_api(method: str, url: str, body: dict = None, headers: dict = None) -> dict:
    """Make HTTP requests. Use for calling external APIs."""
    async with httpx.AsyncClient(timeout=30) as client:
        response = await client.request(method, url, json=body, headers=headers)
        return {"status": response.status_code, "body": response.json()}

# LangGraph interrupt — pause for human approval
from langgraph.graph import StateGraph
from langgraph.checkpoint.memory import MemorySaver

def agent_node(state):
    # ... agent logic ...
    return state

def action_node(state):
    """Pauses here until human approves."""
    # This node will be interrupted before execution
    action = state["planned_action"]
    execute_action(action)  # Only runs after approval
    return {**state, "action_done": True}

builder = StateGraph(State)
builder.add_node("agent", agent_node)
builder.add_node("action", action_node)
builder.set_entry_point("agent")
builder.add_edge("agent", "action")

checkpointer = MemorySaver()
# interrupt_before pauses BEFORE executing the node
app = builder.compile(
    checkpointer=checkpointer,
    interrupt_before=["action"]  # Pause before action_node
)

config = {"configurable": {"thread_id": "task-1"}}

# Run until interrupt
app.invoke({"task": "Delete all inactive users"}, config)

# Inspect what the agent planned
state = app.get_state(config)
print("Agent plans to:", state.values["planned_action"])

# Human approves — resume
app.invoke(None, config)  # Resume from checkpoint

# LangSmith — best-in-class for LangChain/LangGraph agents
# https://smith.langchain.com

import langsmith
from langsmith import Client

# Auto-tracing — wraps every LLM call, tool call, chain step
os.environ["LANGCHAIN_TRACING_V2"] = "true"
os.environ["LANGCHAIN_API_KEY"] = "ls__..."
os.environ["LANGCHAIN_PROJECT"] = "my-agent"

# Create an evaluation dataset
client = Client()
dataset = client.create_dataset("agent-eval-v1")
client.create_examples(
    inputs=[{"task": "Summarize this article: ..."}],
    outputs=[{"answer": "expected summary..."}],
    dataset_id=dataset.id
)

# Define an evaluator
from langchain.smith import RunEvalConfig
eval_config = RunEvalConfig(
    evaluators=[
        "qa",              # LLM-graded correctness
        "criteria",        # Check specific criteria
        "embedding_distance",  # Semantic similarity
    ]
)

# Run evaluation
results = langsmith.aevaluate(
    run_agent,          # Your agent function
    data=dataset,
    evaluators=[...],
    experiment_prefix="gpt4o-react-v2"
)

import time
import random
from functools import wraps

def retry_with_backoff(max_retries=3, base_delay=1.0, exceptions=(Exception,)):
    """Decorator for exponential backoff retry."""
    def decorator(fn):
        @wraps(fn)
        def wrapper(*args, **kwargs):
            for attempt in range(max_retries):
                try:
                    return fn(*args, **kwargs)
                except exceptions as e:
                    if attempt == max_retries - 1:
                        raise
                    delay = base_delay * (2 ** attempt) + random.uniform(0, 0.5)
                    print(f"[Retry] Attempt {attempt+1} failed: {e}. Retrying in {delay:.1f}s...")
                    time.sleep(delay)
        return wrapper
    return decorator

@retry_with_backoff(max_retries=3, base_delay=1.0)
def call_llm(messages):
    return client.chat.completions.create(model="gpt-4o", messages=messages)

# Tool-level error handling — return errors for LLM to self-correct
def safe_tool_call(fn, *args, **kwargs) -> dict:
    try:
        result = fn(*args, **kwargs)
        return {"success": True, "result": result}
    except ValueError as e:
        return {"success": False, "error": f"Invalid input: {e}", "hint": "Check the parameter format"}
    except TimeoutError:
        return {"success": False, "error": "Tool timed out", "hint": "Try a simpler query"}
    except Exception as e:
        return {"success": False, "error": str(e)}
    # The LLM can read the error and self-correct on the next iteration