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?

# ReAct agent loop (simplified)
def run_agent(task: str, tools: dict, max_steps: int = 10):
    messages = [
        {"role": "system", "content": build_system_prompt(tools)},
        {"role": "user", "content": task}
    ]

    for step in range(max_steps):
        # LLM decides next action
        response = llm_call(messages, tools=tools)

        if response.finish_reason == "stop":
            # LLM produced a final answer (no more tool calls)
            return response.content

        if response.tool_calls:
            # Execute the tool call
            for tool_call in response.tool_calls:
                tool_name = tool_call.function.name
                tool_args = json.loads(tool_call.function.arguments)

                # Add the assistant's tool call to message history
                messages.append({"role": "assistant", "tool_calls": [tool_call]})

                # Execute the tool
                result = tools[tool_name](**tool_args)

                # Add tool result to messages (LLM sees what happened)
                messages.append({
                    "role": "tool",
                    "tool_call_id": tool_call.id,
                    "content": str(result)
                })

    return "Max steps reached without completion"

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)