AI Agents — From Basics to Multi-Agent Systems

import anthropic
import json

client = anthropic.Anthropic()

# The actual function — runs on YOUR machine
def get_weather(city: str) -> dict:
    mock = {
        "Amsterdam": {"temp_c": 14, "condition": "Partly cloudy"},
        "London":    {"temp_c": 11, "condition": "Rainy"},
    }
    return mock.get(city, {"error": "City not found"})


# Tool definition — what the model reads
tools = [{
    "name": "get_weather",
    "description": "Get current weather for a city. Returns temp in Celsius and weather condition.",
    "input_schema": {
        "type": "object",
        "properties": {
            "city": {"type": "string", "description": "City name, e.g. 'Amsterdam'"}
        },
        "required": ["city"]
    }
}]


def run_agent(user_message: str) -> str:
    messages = [{"role": "user", "content": user_message}]

    while True:
        response = client.messages.create(
            model="claude-3-5-sonnet-20241022",
            max_tokens=1024,
            tools=tools,
            messages=messages,
        )

        # Model wants to call a tool
        if response.stop_reason == "tool_use":
            tool_block = next(b for b in response.content if b.type == "tool_use")
            name, args = tool_block.name, tool_block.input

            print(f"  → Tool: {name}({args})")
            result = get_weather(**args) if name == "get_weather" else {"error": "unknown tool"}
            print(f"  ← Result: {result}")

            # Add exchange to conversation and loop
            messages.append({"role": "assistant", "content": response.content})
            messages.append({
                "role": "user",
                "content": [{"type": "tool_result", "tool_use_id": tool_block.id, "content": json.dumps(result)}]
            })

        else:
            # Model gave a final answer — we are done
            return next(b.text for b in response.content if hasattr(b, "text"))


print(run_agent("What is the weather in Amsterdam right now?"))

from langchain_anthropic import ChatAnthropic
from langgraph.prebuilt import create_react_agent
from langchain_community.tools import DuckDuckGoSearchRun
from langchain_core.tools import tool

search = DuckDuckGoSearchRun()

@tool
def save_file(filename: str, content: str) -> str:
    """Save content to a file in the current directory."""
    with open(filename, "w") as f:
        f.write(content)
    return f"Saved {len(content)} chars to {filename}"

model = ChatAnthropic(model="claude-3-5-sonnet-20241022")
agent = create_react_agent(model, tools=[search, save_file])

result = agent.invoke({
    "messages": [("human",
        "Research what MCP (Model Context Protocol) is, write a 300-word summary, "
        "and save it to mcp-summary.txt")]
})
print(result["messages"][-1].content)

from langchain_core.tools import tool

@tool
def handle_writing(topic: str, length: str) -> str:
    """Handle a content writing request — blog post, email, or summary."""
    return f"[Writing handler]: {length} piece about {topic}"

@tool
def handle_code(language: str, task: str) -> str:
    """Handle a code writing or explanation request."""
    return f"[Code handler]: {language} — {task}"

@tool
def handle_research(query: str) -> str:
    """Handle a research or fact-finding request."""
    return f"[Research handler]: {query}"

router = create_react_agent(model, tools=[handle_writing, handle_code, handle_research])

import subprocess

@tool
def run_python(code: str) -> str:
    """
    Execute a Python code snippet and return its stdout output.
    Safe, non-destructive operations only. No file writes or network calls.
    Times out after 10 seconds.
    """
    result = subprocess.run(
        ["python3", "-c", code],
        capture_output=True, text=True, timeout=10
    )
    return result.stdout if result.returncode == 0 else f"Error:\n{result.stderr}"

import json
from pathlib import Path

MEMORY = Path("agent_memory.json")

def save_fact(key: str, value: str):
    """Save a fact to persistent memory."""
    facts = json.loads(MEMORY.read_text()) if MEMORY.exists() else {}
    facts[key] = value
    MEMORY.write_text(json.dumps(facts, indent=2))

def recall_fact(key: str) -> str | None:
    """Recall a saved fact."""
    if not MEMORY.exists():
        return None
    return json.loads(MEMORY.read_text()).get(key)

# Usage examples:
save_fact("user_timezone", "Europe/Amsterdam")
save_fact("user_language", "English")
recall_fact("user_timezone")  # → "Europe/Amsterdam"

# Simplified example — in production use Pinecone, Qdrant, or pgvector
import chromadb

chroma = chromadb.Client()
collection = chroma.create_collection("agent_memory")

def store_memory(text: str, doc_id: str):
    collection.add(documents=[text], ids=[doc_id])

def recall_similar(query: str, n: int = 3) -> list[str]:
    results = collection.query(query_texts=[query], n_results=n)
    return results["documents"][0]

import json, datetime

def log_episode(task: str, steps: list[str], outcome: str, cost_usd: float):
    episode = {
        "timestamp": datetime.datetime.utcnow().isoformat(),
        "task": task,
        "steps": steps,
        "outcome": outcome,
        "cost_usd": cost_usd,
    }
    with open("episodes.jsonl", "a") as f:
        f.write(json.dumps(episode) + "\n")

from typing import TypedDict, Annotated
import operator
from langgraph.graph import StateGraph, START, END
from langchain_anthropic import ChatAnthropic
from langchain_core.messages import HumanMessage, BaseMessage

class PipelineState(TypedDict):
    topic: str
    research: str
    draft: str
    messages: Annotated[list[BaseMessage], operator.add]

model = ChatAnthropic(model="claude-3-5-sonnet-20241022")


def researcher(state: PipelineState) -> PipelineState:
    print("[Researcher] Working...")
    response = model.invoke([HumanMessage(content=
        f"Research '{state['topic']}'. Write structured notes covering key concepts, "
        f"recent developments, and important examples. Be specific and factual."
    )])
    return {"research": response.content}


def writer(state: PipelineState) -> PipelineState:
    print("[Writer] Drafting...")
    response = model.invoke([HumanMessage(content=
        f"Using these research notes, write a 400-600 word blog post about '{state['topic']}'.\n"
        f"Include a clear title, 3-4 paragraphs, and a practical conclusion.\n\n"
        f"Research:\n{state['research']}"
    )])
    return {"draft": response.content}


def editor(state: PipelineState) -> PipelineState:
    print("[Editor] Reviewing...")
    response = model.invoke([HumanMessage(content=
        f"Edit this draft. Check for clarity, accuracy, and tone. "
        f"Return the improved version at the same length.\n\nDraft:\n{state['draft']}"
    )])
    return {"draft": response.content}


# Build the graph
builder = StateGraph(PipelineState)
builder.add_node("researcher", researcher)
builder.add_node("writer", writer)
builder.add_node("editor", editor)
builder.add_edge(START, "researcher")
builder.add_edge("researcher", "writer")
builder.add_edge("writer", "editor")
builder.add_edge("editor", END)

pipeline = builder.compile()

result = pipeline.invoke({
    "topic": "Why MCP changes how AI agents connect to tools",
    "research": "", "draft": "", "messages": []
})
print(result["draft"])

def needs_revision(state: PipelineState) -> str:
    """Return 'researcher' to loop back, or 'end' to finish."""
    if "INSUFFICIENT_DATA" in state["draft"]:
        return "researcher"
    return END

builder.add_conditional_edges("editor", needs_revision, {
    "researcher": "researcher",
    END: END,
})

from langgraph.prebuilt import create_react_agent
from langchain_anthropic import ChatAnthropic
from langchain_core.tools import tool

@tool
def search(query: str) -> str:
    """Search the web and return a summary of results."""
    return f"[Search results for: {query}]"

agent = create_react_agent(
    ChatAnthropic(model="claude-3-5-sonnet-20241022"),
    tools=[search]
)
result = agent.invoke({"messages": [("human", "What is LangGraph?")]})
print(result["messages"][-1].content)

from crewai import Agent, Task, Crew, Process

researcher = Agent(
    role="Research Analyst",
    goal="Find accurate, up-to-date information on any topic",
    backstory="Expert at synthesising information from multiple sources.",
    llm="claude-3-5-sonnet-20241022",
    verbose=True,
)

writer = Agent(
    role="Technical Writer",
    goal="Write clear, engaging content about technical topics",
    backstory="Turns complex ideas into readable prose.",
    llm="claude-3-5-sonnet-20241022",
)

research_task = Task(
    description="Research AI agents. Find 5 key capabilities with examples.",
    expected_output="Structured list of 5 capabilities with real examples.",
    agent=researcher,
)

write_task = Task(
    description="Write a 400-word blog post using the research.",
    expected_output="Complete blog post with title and 3 paragraphs.",
    agent=writer,
)

crew = Crew(
    agents=[researcher, writer],
    tasks=[research_task, write_task],
    process=Process.sequential,
)

result = crew.kickoff()
print(result.raw)

import autogen

config_list = [{"model": "claude-3-5-sonnet-20241022", "api_key": "your-key", "api_type": "anthropic"}]

assistant = autogen.AssistantAgent(
    name="assistant",
    llm_config={"config_list": config_list},
    system_message="Write clean, well-documented Python code.",
)

user_proxy = autogen.UserProxyAgent(
    name="user_proxy",
    human_input_mode="NEVER",
    code_execution_config={"work_dir": "output", "use_docker": False},
    max_consecutive_auto_reply=5,
)

user_proxy.initiate_chat(
    assistant,
    message="Write a function to calculate Fibonacci numbers, then test it with n=10."
)

from agno.agent import Agent
from agno.models.anthropic import Claude
from agno.tools.duckduckgo import DuckDuckGoTools
from agno.tools.file import FileTools

agent = Agent(
    model=Claude(id="claude-3-5-sonnet-20241022"),
    tools=[DuckDuckGoTools(), FileTools()],
    description="Research assistant that finds information and saves summaries.",
    markdown=True,
)

agent.print_response(
    "Search for AI agent frameworks in 2025, summarise the top 3, and save to research.md",
    stream=True,
)

from swarm import Swarm, Agent

client = Swarm()

writer = Agent(name="Writer", instructions="Write clear, structured content on any topic.")

def transfer_to_writer():
    return writer

router = Agent(
    name="Router",
    instructions="Decide if the request needs research or writing, then transfer to the right agent.",
    functions=[transfer_to_writer],
)

response = client.run(
    agent=router,
    messages=[{"role": "user", "content": "Write a short intro to AI agents"}]
)
print(response.messages[-1]["content"])

from smolagents import CodeAgent, DuckDuckGoSearchTool, HfApiModel

agent = CodeAgent(
    tools=[DuckDuckGoSearchTool()],
    model=HfApiModel("Qwen/Qwen2.5-72B-Instruct"),
)

agent.run("Search for Python best practices in 2025 and list the top 5.")

pip install langfuse

export LANGFUSE_SECRET_KEY="sk-lf-..."
export LANGFUSE_PUBLIC_KEY="pk-lf-..."
export LANGFUSE_HOST="http://localhost:3000"

from langfuse.callback import CallbackHandler
from langgraph.prebuilt import create_react_agent
from langchain_anthropic import ChatAnthropic
from langchain_core.tools import tool

langfuse = CallbackHandler()

@tool
def search_web(query: str) -> str:
    """Search the web for information."""
    return f"[Results for: {query}]"

agent = create_react_agent(
    ChatAnthropic(model="claude-3-5-sonnet-20241022"),
    tools=[search_web]
)

result = agent.invoke(
    {"messages": [("human", "What is MCP?")]},
    config={"callbacks": [langfuse]}   # ← one line to add full tracing
)

from langchain_core.callbacks import BaseCallbackHandler

class CostGuard(BaseCallbackHandler):
    def __init__(self, max_usd: float):
        self.max_usd = max_usd
        self.total = 0.0

    def on_llm_end(self, response, **kwargs):
        usage = response.llm_output.get("usage", {})
        # Rough estimate for claude-3.5-sonnet
        cost = (usage.get("input_tokens", 0) * 3 + usage.get("output_tokens", 0) * 15) / 1_000_000
        self.total += cost
        print(f"  Cost so far: ${self.total:.4f}")
        if self.total > self.max_usd:
            raise RuntimeError(f"Budget exceeded: ${self.total:.4f} > ${self.max_usd}")

guard = CostGuard(max_usd=0.50)
# config={"callbacks": [langfuse, guard]}

from langgraph.graph import StateGraph, START, END
from langgraph.checkpoint.memory import MemorySaver
from langgraph.types import interrupt, Command
from typing import TypedDict

class State(TypedDict):
    task: str
    plan: str
    approved: bool
    result: str


def planner(state: State) -> State:
    plan = f"1. Research '{state['task']}'\n2. Write a 300-word post\n3. Save to drafts/"
    print(f"[Planner] Plan created:\n{plan}")
    return {"plan": plan}


def approval_gate(state: State) -> State:
    # interrupt() pauses the graph here and returns control to the caller
    decision = interrupt({
        "message": "Please review the plan and approve or reject:",
        "plan": state["plan"],
    })
    return {"approved": decision.get("approved", False)}


def executor(state: State) -> State:
    if not state["approved"]:
        return {"result": "Cancelled by reviewer."}
    return {"result": f"Completed: {state['plan']}"}


# Build with memory (required for interrupt)
memory = MemorySaver()
builder = StateGraph(State)
builder.add_node("planner", planner)
builder.add_node("approval", approval_gate)
builder.add_node("executor", executor)
builder.add_edge(START, "planner")
builder.add_edge("planner", "approval")
builder.add_edge("approval", "executor")
builder.add_edge("executor", END)
app = builder.compile(checkpointer=memory)

# ── Run 1: starts and pauses at approval_gate ──
thread = {"configurable": {"thread_id": "run-001"}}
for _ in app.stream(
    {"task": "AI agent observability", "plan": "", "approved": False, "result": ""},
    config=thread
):
    pass

print("\n--- Agent paused. Human reviewing... ---")

# ── Run 2: resume with human decision ──
for event in app.stream(Command(resume={"approved": True}), config=thread):
    if "executor" in event:
        print("Result:", event["executor"]["result"])

"""
research_pipeline.py — Complete multi-agent research and writing pipeline.

Install:
    pip install langgraph langchain-anthropic langfuse duckduckgo-search

Run:
    python research_pipeline.py
"""

import operator
from pathlib import Path
from typing import TypedDict, Annotated

from langchain_anthropic import ChatAnthropic
from langchain_community.tools import DuckDuckGoSearchRun
from langchain_core.messages import HumanMessage, BaseMessage
from langgraph.checkpoint.memory import MemorySaver
from langgraph.graph import StateGraph, START, END
from langgraph.types import interrupt, Command
from langfuse.callback import CallbackHandler

# ─── State ────────────────────────────────────────────────────────────────────

class State(TypedDict):
    topic: str
    research: str
    draft: str
    approved: bool
    saved_to: str
    messages: Annotated[list[BaseMessage], operator.add]


# ─── Setup ────────────────────────────────────────────────────────────────────

model = ChatAnthropic(model="claude-3-5-sonnet-20241022")
search = DuckDuckGoSearchRun()
langfuse = CallbackHandler()
OUTPUT_DIR = Path.home() / "ai-workspace" / "published"
OUTPUT_DIR.mkdir(parents=True, exist_ok=True)

CALLBACKS = {"callbacks": [langfuse]}


# ─── Nodes ────────────────────────────────────────────────────────────────────

def researcher_node(state: State) -> State:
    print(f"\n[Researcher] Researching: {state['topic']}")
    raw = search.run(state["topic"])
    response = model.invoke([
        HumanMessage(content=
            f"Synthesise these search results about '{state['topic']}' into structured notes:\n\n{raw}\n\n"
            f"Cover: key concepts, recent developments, practical examples, limitations."
        )
    ], config=CALLBACKS)
    return {"research": response.content}


def writer_node(state: State) -> State:
    print("[Writer] Writing draft...")
    response = model.invoke([
        HumanMessage(content=
            f"Write a 500-word blog post about '{state['topic']}' using these notes:\n\n{state['research']}\n\n"
            f"Requirements: specific title, hook opening, 3-4 body paragraphs, actionable conclusion. "
            f"Tone: practical and honest, no hype words."
        )
    ], config=CALLBACKS)
    return {"draft": response.content}


def approval_node(state: State) -> State:
    print("\n[Approval Gate] Pausing for human review...")
    print("─" * 50)
    print(state["draft"][:400], "\n...[truncated]")
    print("─" * 50)

    decision = interrupt({
        "message": "Review the draft above. Approve to publish?",
        "preview": state["draft"][:400],
    })
    return {"approved": decision.get("approved", False)}


def publisher_node(state: State) -> State:
    slug = state["topic"].lower().replace(" ", "-")[:50]
    path = OUTPUT_DIR / f"{slug}.md"
    path.write_text(state["draft"], encoding="utf-8")
    status = "published" if state["approved"] else "rejected"
    print(f"[Publisher] Saved ({status}) → {path}")
    return {"saved_to": str(path)}


# ─── Graph ────────────────────────────────────────────────────────────────────

def build_pipeline():
    memory = MemorySaver()
    g = StateGraph(State)
    g.add_node("researcher", researcher_node)
    g.add_node("writer", writer_node)
    g.add_node("approval", approval_node)
    g.add_node("publisher", publisher_node)
    g.add_edge(START, "researcher")
    g.add_edge("researcher", "writer")
    g.add_edge("writer", "approval")
    g.add_edge("approval", "publisher")
    g.add_edge("publisher", END)
    return g.compile(checkpointer=memory)


# ─── Run ──────────────────────────────────────────────────────────────────────

def run(topic: str):
    app = build_pipeline()
    thread = {"configurable": {"thread_id": f"run-{topic[:20].replace(' ', '-')}"}}
    initial = {"topic": topic, "research": "", "draft": "", "approved": False, "saved_to": "", "messages": []}

    print(f"\n{'='*50}")
    print(f"Topic: {topic}")
    print(f"{'='*50}")

    for _ in app.stream(initial, config=thread):
        pass  # progress is printed inside nodes

    # Simulate human review (set approved=False to reject)
    print("\n>>> Human reviewing (auto-approving for demo)...")
    for event in app.stream(Command(resume={"approved": True}), config=thread):
        pass

    print("\n>>> Done. Check Langfuse for full traces.")


if __name__ == "__main__":
    run("How AI agents work and why they matter in 2025")

export ANTHROPIC_API_KEY="your-key"
export LANGFUSE_SECRET_KEY="sk-lf-..."
export LANGFUSE_PUBLIC_KEY="pk-lf-..."

python research_pipeline.py