Agent Architectures & Reasoning

from anthropic import Anthropic
import json

client = Anthropic()

SYSTEM_PROMPT = """You are a helpful research agent. For each task, think step-by-step.

Use this format:
Thought: <your reasoning about what to do next>
Action: <tool_name>
Action Input: <json input for the tool>

After seeing an Observation, continue with another Thought/Action or give a Final Answer:
Final Answer: <your complete answer to the user>"""

tools = [
    {
        "name": "search_web",
        "description": "Search the internet for current information",
        "input_schema": {
            "type": "object",
            "properties": {
                "query": {"type": "string", "description": "Search query"}
            },
            "required": ["query"]
        }
    },
    {
        "name": "calculate",
        "description": "Perform mathematical calculations",
        "input_schema": {
            "type": "object",
            "properties": {
                "expression": {"type": "string", "description": "Math expression to evaluate"}
            },
            "required": ["expression"]
        }
    }
]

def execute_tool(name: str, inputs: dict) -> str:
    """Mock tool execution — replace with real implementations."""
    if name == "search_web":
        return f"[Search results for: {inputs['query']}] Found 10 relevant results..."
    elif name == "calculate":
        try:
            return str(eval(inputs["expression"]))
        except Exception as e:
            return f"Error: {e}"
    return "Tool not found"

def react_agent(user_query: str, max_steps: int = 6) -> str:
    messages = [{"role": "user", "content": user_query}]

    for step in range(max_steps):
        response = client.messages.create(
            model="claude-sonnet-4-6",
            max_tokens=2048,
            system=SYSTEM_PROMPT,
            tools=tools,
            messages=messages,
        )

        # If model gave a final text answer
        if response.stop_reason == "end_turn":
            for block in response.content:
                if hasattr(block, "text"):
                    return block.text

        # Handle tool calls
        messages.append({"role": "assistant", "content": response.content})
        tool_results = []
        for block in response.content:
            if block.type == "tool_use":
                result = execute_tool(block.name, block.input)
                print(f"  Tool: {block.name}({block.input}) -> {result[:80]}...")
                tool_results.append({
                    "type": "tool_result",
                    "tool_use_id": block.id,
                    "content": result,
                })

        if tool_results:
            messages.append({"role": "user", "content": tool_results})

    return "Agent reached maximum steps without completing the task."

# Usage
result = react_agent("What is 15% of the GDP of France in 2023?")
print(result)

# Good: specific, typed, with clear descriptions
tools = [
    {
        "name": "get_stock_price",
        "description": (
            "Retrieve the current stock price and 24h change for a given ticker symbol. "
            "Use this when the user asks about stock prices, market data, or investment info."
        ),
        "input_schema": {
            "type": "object",
            "properties": {
                "ticker": {
                    "type": "string",
                    "description": "Stock ticker symbol (e.g. AAPL, TSLA, MSFT)"
                },
                "currency": {
                    "type": "string",
                    "enum": ["USD", "EUR", "GBP"],
                    "description": "Currency for the price. Defaults to USD.",
                    "default": "USD"
                }
            },
            "required": ["ticker"]
        }
    },
    {
        "name": "get_company_info",
        "description": (
            "Get fundamental information about a publicly traded company: "
            "sector, market cap, P/E ratio, and recent earnings."
        ),
        "input_schema": {
            "type": "object",
            "properties": {
                "company_name_or_ticker": {
                    "type": "string",
                    "description": "Company name (e.g. 'Apple') or ticker (e.g. 'AAPL')"
                }
            },
            "required": ["company_name_or_ticker"]
        }
    }
]

# Bad: vague, overloaded tool (avoid this)
bad_tool = {
    "name": "get_data",
    "description": "Get some data",
    "input_schema": {
        "type": "object",
        "properties": {
            "query": {"type": "string"}
        }
    }
}

from anthropic import Anthropic
import chromadb

client = Anthropic()
chroma = chromadb.Client()
collection = chroma.get_or_create_collection("agent_memory")

def remember(key: str, content: str) -> None:
    """Store a memory in the vector database."""
    collection.add(
        documents=[content],
        ids=[key],
        metadatas=[{"key": key}]
    )

def recall(query: str, n_results: int = 3) -> list[str]:
    """Retrieve relevant memories by semantic similarity."""
    results = collection.query(query_texts=[query], n_results=n_results)
    return results["documents"][0] if results["documents"] else []

def agent_with_memory(user_query: str) -> str:
    # Retrieve relevant memories
    memories = recall(user_query)
    memory_context = ""
    if memories:
        memory_context = "\n\nRelevant past context:\n" + "\n".join(
            f"- {m}" for m in memories
        )

    response = client.messages.create(
        model="claude-sonnet-4-6",
        max_tokens=1024,
        system=f"You are a helpful assistant with memory.{memory_context}",
        messages=[{"role": "user", "content": user_query}],
    )

    answer = response.content[0].text

    # Store this interaction as a memory
    remember(
        key=f"interaction_{hash(user_query)}",
        content=f"Q: {user_query}\nA: {answer[:200]}"
    )

    return answer