AI Agents

LLM chatbot (not an agent):
  User: "What's the weather in Mumbai?"
  LLM:  "I don't have real-time data, but Mumbai typically..."
  → One turn. No action taken.

AI agent:
  User: "Book me a flight to Mumbai for next Friday under ₹8,000"
  Agent: 
    1. Search flights for next Friday Mumbai
    2. Filter by price < ₹8,000
    3. Check seat availability
    4. Confirm with user
    5. Complete booking
  → Multi-step. Real actions. Real results.

Loop until task complete:
  1. THINK (Reason): given the current state, what should I do next?
  2. ACT: call a tool (search, code, API, etc.)
  3. OBSERVE: receive the tool's result
  4. THINK again: does this result complete the task? What's next?

import re
from typing import Dict, Any, Callable

# --- 1. Define Mock Tools ---
def get_user_balance(username: str) -> str:
    """Get the account balance for a given username."""
    balances = {"Alice": "150", "Bob": "250"}
    return balances.get(username, "User not found")

def calculator(expression: str) -> str:
    """Evaluate a mathematical expression."""
    clean_expr = re.sub(r"[^0-9\+\-\*\/\(\)\. ]", "", expression)
    try:
        return str(eval(clean_expr))
    except Exception as e:
        return f"Error evaluating expression: {e}"

# Tool registry
tools: Dict[str, Callable[[str], str]] = {
    "get_user_balance": get_user_balance,
    "calculator": calculator
}

# --- 2. Mock LLM Engine for Simulation ---
class SimulatedLLM:
    """Simulates LLM completions for a ReAct loop to keep the code runnable without API keys."""
    def __init__(self):
        self.step = 0

    def get_completion(self, prompt: str) -> str:
        self.step += 1
        if self.step == 1:
            return (
                "Thought: I need to find Alice's balance first before performing any calculations.\n"
                "Action: get_user_balance: Alice"
            )
        elif self.step == 2:
            return (
                "Thought: I see that Alice's balance is 150. Now I need to double this amount using the calculator.\n"
                "Action: calculator: 150 * 2"
            )
        elif self.step == 3:
            return (
                "Thought: The calculation output is 300. I have completed the task.\n"
                "Final Answer: Alice's doubled balance is 300."
            )
        return "Final Answer: Done."

# --- 3. ReAct Orchestrator Loop ---
def run_react_agent(task: str, max_steps: int = 5) -> str:
    llm = SimulatedLLM()
    # The agent's working memory / notepad
    notepad = f"Task: {task}\n"
    
    print(f"=== Starting Agentic Loop for Task: '{task}' ===\n")
    
    for step in range(1, max_steps + 1):
        print(f"--- Step {step} ---")
        
        # Get completion from the simulated LLM
        response = llm.get_completion(notepad)
        print(response)
        
        # Add LLM response to notepad
        notepad += response + "\n"
        
        # Check if the model decided it has completed the task
        if "Final Answer:" in response:
            final_match = re.search(r"Final Answer:\s*(.*)", response)
            return final_match.group(1) if final_match else response
            
        # Parse Thought and Action from response
        action_match = re.search(r"Action:\s*(\w+):\s*(.*)", response)
        if not action_match:
            print("Error: Could not parse Action. Stopping loop.")
            break
            
        tool_name = action_match.group(1)
        tool_args = action_match.group(2).strip()
        
        # Execute tool if it exists
        if tool_name in tools:
            print(f"\n[Executing Tool] Calling '{tool_name}' with arguments: '{tool_args}'")
            try:
                observation = tools[tool_name](tool_args)
            except Exception as e:
                observation = f"Error executing tool: {e}"
        else:
            observation = f"Error: Tool '{tool_name}' not found."
            
        print(f"[Observation] {observation}\n")
        
        # Append observation back to notepad for next LLM turn
        notepad += f"Observation: {observation}\n"
        
    return "Agent failed to complete task within step limit."

# Run the ReAct agent
if __name__ == "__main__":
    task = "Double the balance of Alice's account"
    final_output = run_react_agent(task)
    print(f"\n=== Final Agent Output ===\n{final_output}")

import json
from openai import OpenAI

client = OpenAI()

# --- Define tools ---
def search_web(query: str) -> str:
    """Search the web and return relevant results."""
    # Real: use SerpAPI, Tavily, Bing Search API
    return f"Search results for '{query}': [result1, result2...]"

def run_python(code: str) -> str:
    """Execute Python code and return the output."""
    import io, sys, contextlib
    buf = io.StringIO()
    with contextlib.redirect_stdout(buf):
        try:
            exec(code, {})
        except Exception as e:
            return f"Error: {e}"
    return buf.getvalue()

def read_file(path: str) -> str:
    """Read a file and return its contents."""
    try:
        with open(path) as f:
            return f.read()
    except FileNotFoundError:
        return f"File not found: {path}"

# --- OpenAI tool schema (what the LLM sees) ---
tools_schema = [
    {
        "type": "function",
        "function": {
            "name": "search_web",
            "description": "Search the web for current information",
            "parameters": {
                "type": "object",
                "properties": {
                    "query": {"type": "string", "description": "The search query"}
                },
                "required": ["query"]
            }
        }
    },
    {
        "type": "function",
        "function": {
            "name": "run_python",
            "description": "Execute Python code and return the output. Use for calculations, data processing, or any computational task.",
            "parameters": {
                "type": "object",
                "properties": {
                    "code": {"type": "string", "description": "Python code to execute"}
                },
                "required": ["code"]
            }
        }
    }
]

# Tool registry — maps name → function
tool_registry = {
    "search_web": search_web,
    "run_python": run_python,
    "read_file": read_file,
}

class AgentMemory:
    """Three types of memory for AI agents."""

    # 1. In-context memory — everything in the current conversation
    #    Limited by context window. Lost when conversation ends.
    self.messages = []

    # 2. Episodic memory — summaries of past conversations
    #    Stored externally (database), retrieved when relevant
    def save_episode(self, summary: str):
        db.save({"type": "episode", "summary": summary, "timestamp": now()})

    def recall_episodes(self, query: str) -> list[str]:
        # Retrieve semantically similar past episodes
        return vector_search(query, collection="episodes", top_k=3)

    # 3. Semantic memory — learned facts, user preferences
    def remember_fact(self, key: str, value: str):
        db.save({"type": "fact", "key": key, "value": value, "user_id": self.user_id})

    def recall_fact(self, key: str) -> str:
        return db.get({"type": "fact", "key": key, "user_id": self.user_id})

# Example: persistent user preferences
class PersonalAssistantAgent:
    def __init__(self, user_id: str):
        self.memory = AgentMemory(user_id)

    def chat(self, message: str) -> str:
        # Retrieve relevant past context
        past_context = self.memory.recall_episodes(message)
        preferences = self.memory.recall_facts()

        system = f"""You are a personal assistant.
User preferences: {preferences}
Relevant past context: {past_context}"""

        response = run_agent(message, system_prompt=system)

        # Save this episode to memory
        self.memory.save_episode(f"User asked: {message}. Agent responded: {response[:200]}")
        return response

# Orchestrator pattern — one coordinator, multiple specialists

class ResearchAgent:
    """Specialized in web search and summarization"""
    tools = ["search_web", "summarize"]

class CodeAgent:
    """Specialized in writing and running code"""
    tools = ["run_python", "read_file", "write_file"]

class WriterAgent:
    """Specialized in drafting reports and documents"""
    tools = ["write_to_doc", "format_markdown"]

class OrchestratorAgent:
    """Breaks tasks, delegates to specialists, assembles results"""

    def run(self, task: str) -> str:
        # Break task into sub-tasks
        plan = self.plan(task)

        results = {}
        for step in plan:
            if step.type == "research":
                results[step.id] = ResearchAgent().run(step.query)
            elif step.type == "code":
                results[step.id] = CodeAgent().run(step.code_task)
            elif step.type == "write":
                results[step.id] = WriterAgent().run(step.content, results)

        return self.assemble(task, results)

# LangGraph example — stateful agent with tools
from langgraph.prebuilt import create_react_agent
from langchain_openai import ChatOpenAI

llm = ChatOpenAI(model="gpt-4o")
tools = [search_web_tool, run_python_tool]  # LangChain tool objects

agent = create_react_agent(llm, tools)

result = agent.invoke({
    "messages": [{"role": "user", "content": "Research the top 5 vector databases and compare their features"}]
})
print(result["messages"][-1].content)