Building Agents with Frameworks

import anthropic
import json
from datetime import datetime
 
client = anthropic.Anthropic()
 
# ── TOOL DEFINITIONS ─────────────────────────────────────────────
 
tools = [
    {
        "name": "get_weather",
        "description": "Get the current weather for a city. Returns temperature in Celsius and conditions.",
        "input_schema": {
            "type": "object",
            "properties": {
                "city": {
                    "type": "string",
                    "description": "The city name, e.g. 'San Francisco' or 'Tokyo'"
                },
                "units": {
                    "type": "string",
                    "enum": ["celsius", "fahrenheit"],
                    "description": "Temperature unit. Defaults to celsius."
                }
            },
            "required": ["city"]
        }
    },
    {
        "name": "get_current_time",
        "description": "Get the current date and time.",
        "input_schema": {
            "type": "object",
            "properties": {},
            "required": []
        }
    }
]
 
 
# ── TOOL IMPLEMENTATIONS ──────────────────────────────────────────
 
def get_weather(city: str, units: str = "celsius") -> dict:
    """Simulated weather API — replace with real API call in production."""
    mock_data = {
        "san francisco": {"temp_c": 16, "conditions": "Partly cloudy", "humidity": 78},
        "tokyo":         {"temp_c": 28, "conditions": "Sunny",          "humidity": 65},
        "london":        {"temp_c": 12, "conditions": "Overcast",       "humidity": 85},
    }
    data = mock_data.get(city.lower(), {"temp_c": 20, "conditions": "Unknown", "humidity": 60})
    
    if units == "fahrenheit":
        data["temp"] = (data["temp_c"] * 9/5) + 32
        data["unit"] = "F"
    else:
        data["temp"] = data["temp_c"]
        data["unit"] = "C"
    
    return {"city": city, "temperature": f"{data['temp']}{data['unit']}", 
            "conditions": data["conditions"], "humidity": f"{data['humidity']}%"}
 
def get_current_time() -> dict:
    now = datetime.now()
    return {"datetime": now.strftime("%Y-%m-%d %H:%M:%S"), "timezone": "local"}
 
TOOL_REGISTRY = {
    "get_weather": get_weather,
    "get_current_time": get_current_time,
}
 
 
# ── AGENT LOOP ────────────────────────────────────────────────────
 
def run_agent(user_message: str) -> str:
    """Run a tool-use agent loop until the model returns a final answer."""
    
    messages = [{"role": "user", "content": user_message}]
    
    while True:
        response = client.messages.create(
            model="claude-haiku-4-5",
            max_tokens=1024,
            tools=tools,
            messages=messages
        )
        
        # Case 1: Model is done — return the final text response
        if response.stop_reason == "end_turn":
            for block in response.content:
                if hasattr(block, "text"):
                    return block.text
        
        # Case 2: Model wants to call a tool
        elif response.stop_reason == "tool_use":
            # Add the model's (tool-requesting) response to history
            messages.append({"role": "assistant", "content": response.content})
            
            # Execute each requested tool call
            tool_results = []
            for block in response.content:
                if block.type == "tool_use":
                    tool_name = block.name
                    tool_args = block.input
                    
                    print(f"  [Calling tool: {tool_name}({json.dumps(tool_args)})]")
                    
                    fn = TOOL_REGISTRY.get(tool_name)
                    if fn:
                        result = fn(**tool_args)
                        result_str = json.dumps(result)
                    else:
                        result_str = f"Error: Tool '{tool_name}' not found."
                    
                    tool_results.append({
                        "type": "tool_result",
                        "tool_use_id": block.id,
                        "content": result_str
                    })
            
            # Add tool results and loop back to get the model's next response
            messages.append({"role": "user", "content": tool_results})
        
        else:
            # Unexpected stop reason
            return f"Agent stopped unexpectedly: {response.stop_reason}"
 
 
# Test it
result = run_agent("What's the weather like in Tokyo right now, and what time is it?")
print(f"\nFinal answer:\n{result}")

pip install crewai crewai-tools

from crewai import Agent, Task, Crew, Process
from crewai_tools import SerperDevTool  # Web search tool
import os
 
# Set your API keys
os.environ["ANTHROPIC_API_KEY"] = "your-key"
os.environ["SERPER_API_KEY"] = "your-key"  # serper.dev for web search
 
search_tool = SerperDevTool()
 
# ── DEFINE AGENTS ────────────────────────────────────────────────
 
researcher = Agent(
    role="Senior Research Analyst",
    goal="Find and synthesize accurate, up-to-date information on any given topic",
    backstory="""You are a meticulous research analyst with 15 years of experience 
    in technology journalism. You excel at finding credible sources, separating 
    signal from noise, and presenting findings in a clear, structured format.
    You always cite your sources.""",
    tools=[search_tool],
    verbose=True,
    llm="claude-haiku-4-5"   # Use cheap model for research
)
 
writer = Agent(
    role="Content Strategist and Writer",
    goal="Transform research findings into engaging, accurate content for a technical audience",
    backstory="""You are a skilled technical writer who has written for publications 
    like Wired, MIT Technology Review, and The Verge. You turn complex topics into 
    clear, engaging prose without sacrificing accuracy. You write with authority.""",
    verbose=True,
    llm="claude-sonnet-4-5"  # Better model for the actual writing
)
 
# ── DEFINE TASKS ─────────────────────────────────────────────────
 
research_task = Task(
    description="""Research the current state of {topic}.
    
    Your research should cover:
    1. What it is and how it works (technically accurate, but accessible)
    2. Current real-world applications (with specific examples and companies)
    3. Key limitations and challenges
    4. Where things are heading in the next 2 years
    
    Find at least 3 credible sources. Note publication dates — prefer recent.""",
    expected_output="""A structured research brief with:
    - Executive summary (3-4 sentences)
    - Key findings (bulleted, with sources)
    - Limitations and challenges
    - Future outlook
    - Sources (title, URL, date)""",
    agent=researcher
)
 
writing_task = Task(
    description="""Using the research brief provided, write a compelling blog post about {topic}.
    
    Requirements:
    - Length: 600-800 words
    - Audience: software engineers curious about AI
    - Tone: authoritative but accessible, not hype-y
    - Structure: hook → context → substance → implications → conclusion
    - Include 1-2 concrete examples that a developer can relate to
    - No buzzwords: avoid 'revolutionary', 'game-changing', 'unprecedented'""",
    expected_output="""A complete, publication-ready blog post with:
    - Compelling headline
    - Full body (600-800 words)
    - All claims grounded in the research brief""",
    agent=writer,
    context=[research_task]  # Writer receives researcher's output
)
 
# ── ASSEMBLE AND RUN THE CREW ─────────────────────────────────────
 
crew = Crew(
    agents=[researcher, writer],
    tasks=[research_task, writing_task],
    process=Process.sequential,  # Research first, then writing
    verbose=True
)
 
result = crew.kickoff(inputs={"topic": "AI agents in production"})
print(result.raw)

import anthropic
 
client = anthropic.Anthropic()
 
def reflection_agent(
    task: str,
    max_iterations: int = 3,
    quality_threshold: float = 8.0  # out of 10
) -> str:
    """
    Generate → critique → improve loop.
    Returns the best output found within max_iterations.
    """
    
    # Step 1: Generate initial draft
    print(f"=== Generating initial draft ===")
    draft_response = client.messages.create(
        model="claude-sonnet-4-5",
        max_tokens=2048,
        system="You are an expert software engineer. Complete the given task thoroughly.",
        messages=[{"role": "user", "content": task}]
    )
    current_draft = draft_response.content[0].text
    print(f"Draft generated ({len(current_draft)} chars)")
    
    for iteration in range(max_iterations):
        # Step 2: Critique the draft
        print(f"\n=== Critique round {iteration + 1} ===")
        critique_response = client.messages.create(
            model="claude-sonnet-4-5",
            max_tokens=1024,
            system="""You are a strict technical reviewer. Your job is to critique code and writing.
            
Be specific about problems. Focus on:
- Correctness: will it actually work? Edge cases handled?
- Completeness: does it fully address the task?
- Quality: clarity, efficiency, best practices
 
End your critique with:
SCORE: [0-10]
VERDICT: [APPROVE if score >= 8, REVISE if score < 8]""",
            messages=[
                {"role": "user", "content": f"Original task:\n{task}\n\nDraft to review:\n{current_draft}"}
            ]
        )
        
        critique = critique_response.content[0].text
        print(f"Critique:\n{critique}")
        
        # Parse score and verdict
        score_line = [l for l in critique.split("\n") if l.startswith("SCORE:")]
        verdict_line = [l for l in critique.split("\n") if l.startswith("VERDICT:")]
        
        if score_line and verdict_line:
            score = float(score_line[0].split(":")[1].strip())
            verdict = verdict_line[0].split(":")[1].strip()
            
            if verdict == "APPROVE" or score >= quality_threshold:
                print(f"\n Quality threshold met (score: {score}/10). Done.")
                return current_draft
        
        if iteration == max_iterations - 1:
            print(f"\nMax iterations reached. Returning best draft.")
            return current_draft
        
        # Step 3: Improve based on critique
        print(f"\n=== Improving based on critique ===")
        improve_response = client.messages.create(
            model="claude-sonnet-4-5",
            max_tokens=2048,
            system="You are an expert software engineer. Revise the draft based on reviewer feedback.",
            messages=[
                {"role": "user", "content": f"""Original task:
{task}
 
Current draft:
{current_draft}
 
Reviewer feedback:
{critique}
 
Produce an improved version that addresses all the reviewer's concerns."""}
            ]
        )
        
        current_draft = improve_response.content[0].text
        print(f"Draft improved ({len(current_draft)} chars)")
    
    return current_draft
 
 
# Test: code review task
result = reflection_agent(
    task="""Write a Python function that safely reads a JSON file and returns the parsed 
data. Handle all common failure cases gracefully. Include type hints and docstring.""",
    max_iterations=3
)
 
print(f"\n=== FINAL OUTPUT ===\n{result}")

from typing import TypedDict, Annotated
from langgraph.graph import StateGraph, END
from langgraph.prebuilt import ToolNode
import anthropic
 
# Define the state schema — what gets passed between nodes
class AgentState(TypedDict):
    messages: list
    iteration_count: int
 
# Define tools (same format as raw API)
tools = [
    {
        "name": "web_search",
        "description": "Search the web for current information",
        "input_schema": {
            "type": "object",
            "properties": {
                "query": {"type": "string", "description": "Search query"}
            },
            "required": ["query"]
        }
    }
]
 
client = anthropic.Anthropic()
 
# Node: LLM call
def researcher_node(state: AgentState) -> AgentState:
    response = client.messages.create(
        model="claude-haiku-4-5",
        max_tokens=1024,
        tools=tools,
        messages=state["messages"]
    )
    
    state["messages"].append({"role": "assistant", "content": response.content})
    state["_stop_reason"] = response.stop_reason
    state["iteration_count"] = state.get("iteration_count", 0) + 1
    
    return state
 
# Node: tool execution
def tool_node(state: AgentState) -> AgentState:
    last_message = state["messages"][-1]
    tool_results = []
    
    for block in last_message["content"]:
        if hasattr(block, "type") and block.type == "tool_use":
            # Execute the tool (simplified — use your actual implementations)
            result = f"Search results for '{block.input['query']}': [mock results]"
            tool_results.append({
                "type": "tool_result",
                "tool_use_id": block.id,
                "content": result
            })
    
    state["messages"].append({"role": "user", "content": tool_results})
    return state
 
# Routing function: decide which node comes next
def router(state: AgentState) -> str:
    if state.get("_stop_reason") == "tool_use":
        return "tools"
    if state.get("iteration_count", 0) >= 5:
        return END  # Safety limit
    return END
 
# Build the graph
workflow = StateGraph(AgentState)
workflow.add_node("researcher", researcher_node)
workflow.add_node("tools", tool_node)
 
workflow.set_entry_point("researcher")
workflow.add_conditional_edges("researcher", router, {
    "tools": "tools",
    END: END
})
workflow.add_edge("tools", "researcher")  # After tools, go back to researcher
 
graph = workflow.compile()
 
# Run the graph
initial_state = {
    "messages": [{"role": "user", "content": "What are the latest developments in AI agents?"}],
    "iteration_count": 0
}
 
result = graph.invoke(initial_state)