[Simple LLM Call]
Question → LLM → Answer
(One-shot, no memory, no tools)
[Chatbot]
Conversation → LLM → Conversation
(Context, memory, no tools)
[AI Agent]
Goal → [Plan → Use Tools → Review Result → Re-plan] → Goal Achieved
(Autonomous, memory, tools, loops)

┌─────────────────────────────────────────────────┐
│                   AI Agent                        │
│                                                   │
│  ┌──────────┐  ┌──────────┐  ┌────────────────┐  │
│  │Perception│  │ Reasoning│  │  Memory System  │  │
│  │  Layer   │→ │  Engine  │←→│                │  │
│  └──────────┘  └─────┬────┘  └────────────────┘  │
│                      │                            │
│               ┌──────▼──────┐                     │
│               │  Tool Layer  │                     │
│               └─────────────┘                     │
└─────────────────────────────────────────────────┘

# Perception layer: handle diverse input types
def perceive(input_data: dict) -> str:
    """
    Process input based on type and convert
    into text the agent can understand.
    """
    if input_data["type"] == "file":
        return parse_file(input_data["path"])
    elif input_data["type"] == "api_response":
        return json.dumps(input_data["data"])
    elif input_data["type"] == "text":
        return input_data["content"]
    else:
        return str(input_data)

# Reasoning engine: LLM decides the next action
def reason(context: str, available_tools: list) -> dict:
    """
    Analyze current situation and decide next action.
    Returns: {"action": "tool_name", "input": "...", "reasoning": "..."}
    """
    prompt = f"""
Current situation: {context}
Available tools: {[t['name'] for t in available_tools]}
Decide one of the following:
1. Use a specific tool → {{"action": "use_tool", "tool": "tool_name", "input": "input_value"}}
2. Generate final answer → {{"action": "final_answer", "content": "answer"}}
Respond with JSON only.
"""
    response = llm.invoke(prompt)
    return json.loads(response.content)

# Memory system: 3 types
class AgentMemory:
    def __init__(self):
        # Short-term: current session conversation
        self.short_term = []
        # Long-term: facts persisted across sessions
        self.long_term = {}
        # Episodic: past task results
        self.episodic = []
    def add_short_term(self, role: str, content: str):
        self.short_term.append({"role": role, "content": content})
        # Compress if too long
        if len(self.short_term) > 20:
            self._compress()
    def save_long_term(self, key: str, value: str):
        self.long_term[key] = {
            "value": value,
            "updated_at": datetime.now().isoformat()
        }
    def get_context(self) -> str:
        """Build context to inject into the reasoning engine"""
        ctx = ""
        if self.long_term:
            ctx += f"[Stored facts]\n{json.dumps(self.long_term)}\n\n"
        ctx += "[Current conversation]\n"
        for msg in self.short_term[-10:]:  # Last 10 only
            ctx += f"{msg['role']}: {msg['content']}\n"
        return ctx

# Tool registration pattern
from langchain.tools import tool
@tool
def search_web(query: str) -> str:
    """Search the web for up-to-date information."""
    # Production: connect Tavily, SerpAPI, etc.
    return f"Search results for: {query}..."
@tool
def execute_python(code: str) -> str:
    """Safely execute Python code and return the result."""
    import subprocess
    result = subprocess.run(
        ["python", "-c", code],
        capture_output=True, text=True, timeout=10
    )
    return result.stdout or result.stderr
@tool
def query_database(sql: str) -> str:
    """Query the database for information."""
    # Production: real DB connection
    return f"Query result for: {sql}"

[ReAct Loop]
Goal input
    ↓
Thought: "To solve this I first need to know X"
    ↓
Action: search_web("information about X")
    ↓
Observation: "Search results: ..."
    ↓
Thought: "Now I need to calculate Y"
    ↓
Action: execute_python("Y calculation code")
    ↓
Observation: "Result: 42"
    ↓
Thought: "I have enough information. I can answer."
    ↓
Final Answer: "..."

[Plan-and-Execute Flow]
Goal: "Write a pricing analysis report for 3 competitors"
    ↓
[Planning Phase]
Plan:
  Step 1: Collect price data from Company A's website
  Step 2: Pull price data from Company B's API
  Step 3: Scrape Company C
  Step 4: Comparative analysis of all 3 datasets
  Step 5: Generate report
    ↓
[Execution Phase: Run each step in sequence]
Execute Step 1 → Execute Step 2 → ... → Execute Step 5
    ↓
Final Report

[Supervisor Pattern]
User request
    ↓
Supervisor Agent
    ├─→ Research Agent   (information gathering specialist)
    ├─→ Analysis Agent   (data processing specialist)
    ├─→ Writing Agent    (document generation specialist)
    └─→ Validation Agent (quality assurance specialist)
    ↓
Integrated output

# react_agent.py
# Install: pip install langchain langchain-anthropic python-dotenv
import os
import json
from dotenv import load_dotenv
from langchain_anthropic import ChatAnthropic
from langchain.tools import tool
from langchain_core.messages import HumanMessage, AIMessage, SystemMessage
load_dotenv()
# ── Tool Definitions ───────────────────────────────────
@tool
def calculator(expression: str) -> str:
    """Performs math calculations. Example: '2 + 2 * 10'"""
    try:
        result = eval(expression, {"__builtins__": {}})
        return f"Result: {result}"
    except Exception as e:
        return f"Calculation error: {str(e)}"
@tool
def get_current_time() -> str:
    """Returns the current date and time."""
    from datetime import datetime
    return f"Current time: {datetime.now().strftime('%Y-%m-%d %H:%M:%S')}"
@tool
def search_knowledge(query: str) -> str:
    """
    Searches the internal knowledge base.
    In production, wire this to a RAG pipeline or search API.
    """
    knowledge_base = {
        "python": "Python holds 26.14% TIOBE share in 2026 — an all-time high for any language.",
        "langgraph": "LangGraph has 126,000+ GitHub stars and is the leading agent framework in 2026.",
        "ai agent": "51% of enterprises run AI agents in production as of 2026.",
    }
    for key, value in knowledge_base.items():
        if key.lower() in query.lower():
            return value
    return f"No information found for '{query}'."
TOOLS = [calculator, get_current_time, search_knowledge]
TOOL_MAP = {t.name: t for t in TOOLS}
# ── ReAct Agent ────────────────────────────────────────
class ReActAgent:
    def __init__(self):
        self.llm = ChatAnthropic(
            model="claude-sonnet-4-20250514",
            api_key=os.getenv("ANTHROPIC_API_KEY"),
            temperature=0,
            max_tokens=2048
        )
        self.max_iterations = 10  # prevent infinite loops
    def _build_system_prompt(self) -> str:
        tool_descriptions = "\n".join(
            f"- {t.name}: {t.description}" for t in TOOLS
        )
        return f"""You are an AI agent that solves problems by using tools.
Available tools:
{tool_descriptions}
Always respond in exactly one of these formats:
When using a tool:
```json
{{"action": "tool_name", "input": "input_value", "thought": "why you chose this tool"}}

{{"action": "final_answer", "content": "your final answer"}}

def run(self, goal: str) -> str:
    """Accept a goal and run the ReAct loop."""
    print(f"\n{'='*60}")
    print(f"🎯 Goal: {goal}")
    print(f"{'='*60}\n")
    messages = [
        SystemMessage(content=self._build_system_prompt()),
        HumanMessage(content=f"Goal: {goal}")
    ]
    for iteration in range(self.max_iterations):
        response = self.llm.invoke(messages)
        raw = response.content.strip()
        # Parse JSON
        try:
            if "```json" in raw:
                raw = raw.split("```json")[1].split("```")[0].strip()
            elif "```" in raw:
                raw = raw.split("```")[1].split("```")[0].strip()
            decision = json.loads(raw)
        except json.JSONDecodeError:
            print(f"⚠️ JSON parse failed. Raw: {raw[:100]}")
            break
        action = decision.get("action")
        thought = decision.get("thought", "")
        # Final answer
        if action == "final_answer":
            final = decision.get("content", "")
            print(f"✅ Final Answer: {final}")
            return final
        # Execute tool
        tool_name = action
        tool_input = decision.get("input", "")
        if thought:
            print(f"💭 Thought: {thought}")
        print(f"🔧 Tool call: {tool_name}({tool_input!r})")
        if tool_name not in TOOL_MAP:
            observation = f"Error: tool '{tool_name}' does not exist."
        else:
            try:
                observation = TOOL_MAP[tool_name].invoke(tool_input)
            except Exception as e:
                observation = f"Tool execution error: {str(e)}"
        print(f"👁️ Observation: {observation}\n")
        messages.append(AIMessage(content=response.content))
        messages.append(HumanMessage(
            content=f"Tool result: {observation}\n\nContinue."
        ))
    return "Max iterations reached. Could not complete the task."

result = agent.run(
    "Check the current time, calculate (2026 - 1991) × 3, "
    "and also tell me about the Python programming language."
)

# ❌ Too many responsibilities
@tool
def do_everything(task: str) -> str:
    """Searches, calculates, and saves files"""
    ...
# ✅ Separation of concerns
@tool
def search_web(query: str) -> str:
    """Searches the web for information."""
    ...
@tool
def save_to_file(filename: str, content: str) -> str:
    """Saves content to a file."""
    ...

@tool
def call_external_api(endpoint: str) -> str:
    """Calls an external API."""
    try:
        response = requests.get(endpoint, timeout=5)
        response.raise_for_status()
        return response.json()
    except requests.Timeout:
        return "Error: API response timed out (5s). Try again later."
    except requests.HTTPError as e:
        return f"Error: HTTP {e.response.status_code} — {e.response.text[:200]}"
    except Exception as e:
        return f"Error: Unexpected problem — {str(e)}"

# ❌ Problem
while True:
    action = agent.decide()
    result = execute(action)
# ✅ Fix: iteration limit + loop detection
MAX_ITER = 10
seen_actions = []
for i in range(MAX_ITER):
    action = agent.decide()
    # Stop if the same action appears 3+ times
    if seen_actions.count(action) >= 3:
        return "Loop detected: same action repeated 3 times. Stopping."
    seen_actions.append(action)
    result = execute(action)

# ✅ Fix: sliding window + summary compression
def get_trimmed_context(messages: list, max_tokens: int = 4000) -> list:
    """Keep only recent messages; compress older ones into a summary."""
    if count_tokens(messages) <= max_tokens:
        return messages
    old_messages = messages[:-10]
    recent_messages = messages[-10:]
    summary = summarize(old_messages)
    return [SystemMessage(content=f"Previous conversation summary: {summary}")] + recent_messages

# ✅ Fix: retry with exponential backoff + fallback path
import time
def execute_with_retry(tool, input_data, max_retries=3):
    for attempt in range(max_retries):
        try:
            return tool.invoke(input_data)
        except Exception as e:
            if attempt < max_retries - 1:
                wait = 2 ** attempt  # 1s, 2s, 4s
                print(f"Retry {attempt + 1}/{max_retries} in {wait}s")
                time.sleep(wait)
            else:
                return f"Tool failed after 3 attempts: {str(e)}. Trying an alternative."

# ✅ Fix: validation layer before execution
def validate_action(action: dict, available_tools: list) -> tuple[bool, str]:
    """Validate the agent's decision before executing it."""
    tool_names = [t.name for t in available_tools]
    if action.get("action") not in tool_names + ["final_answer"]:
        return False, f"Non-existent tool called: '{action.get('action')}'"
    if action.get("action") == "final_answer" and not action.get("content"):
        return False, "Final answer content is empty"
    return True, "OK"

# Customer service agent tools
@tool
def check_order_status(order_id: str) -> str:
    """Look up shipping status and estimated delivery for an order ID."""
    ...
@tool
def process_refund(order_id: str, reason: str) -> str:
    """Process a refund request. Completes within 3–5 business days."""
    ...
@tool
def escalate_to_human(ticket_id: str, priority: str) -> str:
    """Escalate a complex inquiry to a human support agent."""
    ...

@tool
def query_analytics(nl_question: str) -> str:
    """Convert a natural language question to SQL and query the analytics DB.
    Example: 'What were the top 5 best-selling products last month?'"""
    sql = text_to_sql(nl_question)
    return execute_query(sql)
@tool
def generate_chart(data: str, chart_type: str) -> str:
    """Visualize data as a chart and return the file path."""
    ...

@tool
def analyze_code_diff(diff: str) -> str:
    """Analyze code changes for bugs, security vulnerabilities, and performance issues."""
    ...
@tool
def check_test_coverage(file_path: str) -> str:
    """Check test coverage for a file and return uncovered lines."""
    ...