AI Agents — Architecture, LangGraph, Multi-Agent Systems, Tool Use, Planning, Memory & Production Deployment

# Simplified AutoGPT pattern
class AutoGPTAgent:
    def __init__(self, tools, llm):
        self.tools = tools
        self.llm = llm
        self.task_list = []
        self.memory = []
    
    def run(self, goal):
        self.task_list.append({"id": 1, "description": goal, "status": "pending"})
        
        while self.task_list:
            current = next(t for t in self.task_list if t["status"] == "pending")
            print(f"Executing: {current['description']}")
            
            # LLM decides next action
            action = self.plan_next_action(current)
            result = self.execute_action(action)
            
            # Update memory with result
            self.memory.append({"task": current["id"], "result": result})
            
            # Generate new tasks if needed
            new_tasks = self.create_tasks(result, current)
            self.task_list.extend(new_tasks)
            
            current["status"] = "completed"
            
            if self.is_goal_achieved(goal):
                break
        
        return self.summarize()

from crewai import Agent, Task, Crew

# Define specialized agents
researcher = Agent(
    role="Research Analyst",
    goal="Find the latest information on AI agents",
    backstory="Expert in AI research and technology trends",
    verbose=True,
    allow_delegation=False,
)

writer = Agent(
    role="Technical Writer",
    goal="Write a clear, engaging blog post about AI agents",
    backstory="Experienced technical writer specializing in AI topics",
    verbose=True,
    allow_delegation=True,
)

# Define tasks
research_task = Task(
    description="Research the latest developments in AI agents for 2026",
    expected_output="A summary of key trends and technologies",
    agent=researcher,
)

write_task = Task(
    description="Write a 500-word blog post based on the research",
    expected_output="A complete blog post in markdown format",
    agent=writer,
)

# Assemble the crew
crew = Crew(
    agents=[researcher, writer],
    tasks=[research_task, write_task],
    verbose=True,
)

result = crew.kickoff()
print(result)

from typing import TypedDict, Annotated, Sequence
import operator
from langgraph.graph import StateGraph, END
from langchain_openai import ChatOpenAI
from langchain_core.messages import HumanMessage, AIMessage, ToolMessage

# Define state
class AgentState(TypedDict):
    messages: Annotated[Sequence, operator.add]
    next_step: str

# Define tools
def search_web(query: str) -> str:
    """Search the web for recent information."""
    return f"Search results for: {query}"

def calculate(expression: str) -> str:
    """Evaluate a mathematical expression."""
    try:
        return str(eval(expression))
    except:
        return "Error in calculation"

# Create the graph
def create_agent_graph():
    llm = ChatOpenAI(model="gpt-4")
    tools = [search_web, calculate]
    llm_with_tools = llm.bind_tools(tools)
    
    def agent_node(state: AgentState):
        messages = state["messages"]
        response = llm_with_tools.invoke(messages)
        return {"messages": [response], "next_step": "continue"}
    
    def tool_node(state: AgentState):
        last_message = state["messages"][-1]
        tool_calls = last_message.tool_calls
        
        if not tool_calls:
            return {"messages": [], "next_step": "end"}
        
        results = []
        for tc in tool_calls:
            if tc["name"] == "search_web":
                result = search_web(tc["args"]["query"])
            elif tc["name"] == "calculate":
                result = calculate(tc["args"]["expression"])
            else:
                result = "Unknown tool"
            
            results.append(ToolMessage(content=result, tool_call_id=tc["id"]))
        
        return {"messages": results, "next_step": "continue"}
    
    def should_continue(state: AgentState):
        messages = state["messages"]
        last_message = messages[-1]
        
        if hasattr(last_message, "tool_calls") and last_message.tool_calls:
            return "tools"
        return "end"
    
    # Build graph
    graph = StateGraph(AgentState)
    graph.add_node("agent", agent_node)
    graph.add_node("tools", tool_node)
    graph.set_entry_point("agent")
    graph.add_conditional_edges("agent", should_continue, {
        "tools": "tools",
        "end": END,
    })
    graph.add_edge("tools", "agent")
    
    return graph.compile()

# Run the agent
agent = create_agent_graph()
result = agent.invoke({
    "messages": [HumanMessage(
        content="Search for the latest AI news and calculate 2+2"
    )]
})
for msg in result["messages"]:
    print(f"{msg.type}: {msg.content[:100] if msg.content else '[tool call]'}")

# Multi-agent system with LangGraph
class MultiAgentSystem:
    def __init__(self):
        self.orchestrator = self.create_orchestrator()
        self.specialists = {
            "search": SearchAgent(),
            "analyze": AnalysisAgent(),
            "code": CodeAgent(),
            "write": WritingAgent(),
        }
    
    def run(self, task):
        # Orchestrator creates plan with specialist sub-tasks
        plan = self.orchestrator.create_plan(task)
        results = {}
        
        for subtask in plan["subtasks"]:
            specialist_type = subtask["specialist"]
            agent = self.specialists[specialist_type]
            results[subtask["id"]] = agent.execute(subtask)
        
        # Orchestrator synthesizes results
        final = self.orchestrator.synthesize(task, results)
        return final

Thought: I need to find the current weather for London.
Action: search_web(query="London weather 2026")
Observation: London is currently 18°C and partly cloudy.
Thought: The user asked for the weather. I have the answer.
Final: The current weather in London is 18°C and partly cloudy.

def plan_and_execute(task, tools, llm):
    # Step 1: Create plan
    plan_prompt = f"Create a step-by-step plan for: {task}"
    plan = llm.invoke(plan_prompt)
    
    # Step 2: Execute each step
    results = []
    for i, step in enumerate(plan.steps):
        result = execute_step(step, tools)
        results.append(result)
        
        # Step 3: Possibly revise plan based on results
        if should_replan(step, result):
            plan = llm.invoke(f"Revise remaining plan given: {result}")
    
    # Step 4: Synthesize final output
    return llm.invoke(f"Synthesize results: {results}")

# Agent with long-term memory using vector store
class MemoryAgent:
    def __init__(self, vector_store):
        self.short_term = []
        self.long_term = vector_store
    
    def retrieve_relevant_memories(self, query):
        results = self.long_term.similarity_search(query, k=3)
        return [r.page_content for r in results]
    
    def store_memory(self, content, metadata=None):
        self.long_term.add_texts([content], metadatas=[metadata])
    
    def process(self, user_input):
        # Retrieve relevant past memories
        memories = self.retrieve_relevant_memories(user_input)
        context = "\n".join(memories)
        
        # Build prompt with memories + conversation
        prompt = f"Relevant context:\n{context}\n\nUser: {user_input}"
        response = llm.invoke(prompt)
        
        # Store this interaction
        self.short_term.append({"user": user_input, "assistant": response})
        self.store_memory(f"User asked: {user_input}. Response: {response}")
        
        return response

from langchain_core.tools import tool

@tool
def get_stock_price(symbol: str) -> str:
    """Get the current stock price for a given symbol.
    
    Args:
        symbol: Stock ticker symbol (e.g., AAPL, GOOGL)
    """
    import yfinance as yf
    stock = yf.Ticker(symbol)
    price = stock.info.get("currentPrice", "N/A")
    return f"{symbol}: ${price}"

@tool
def send_email(to: str, subject: str, body: str) -> str:
    """Send an email to the specified recipient.
    
    Args:
        to: Recipient email address
        subject: Email subject line
        body: Email body content
    """
    # Email sending logic here
    return f"Email sent to {to} with subject: {subject}"

# Register tools with agent
tools = [get_stock_price, send_email]
agent = create_agent_with_tools(tools)
agent.run("What's Apple's stock price and email it to me")

class ResilientAgent:
    def execute_with_retry(self, tool_call, max_retries=3):
        for attempt in range(max_retries):
            try:
                return tool_call()
            except ToolExecutionError as e:
                print(f"Attempt {attempt + 1} failed: {e}")
                if attempt == max_retries - 1:
                    return f"Failed after {max_retries} attempts: {e}"
                time.sleep(2 ** attempt)  # Exponential backoff
    
    def handle_invalid_action(self, llm_output):
        """If LLM produces invalid JSON, ask it to fix it."""
        correction_prompt = f"Your previous output was invalid: {llm_output}. Please output a valid tool call."
        return llm.invoke(correction_prompt)

# Production agent monitoring
import logging
import time

class MonitoredAgent:
    def __init__(self):
        self.logger = logging.getLogger("agent")
        self.step_count = 0
        self.max_steps = 25
    
    def process(self, user_input):
        start_time = time.time()
        self.logger.info(f"Starting agent run. Input: {user_input[:50]}")
        
        state = {"input": user_input, "steps": [], "output": None}
        
        while self.step_count < self.max_steps:
            step_start = time.time()
            action = self.plan(state)
            
            if action["type"] == "final":
                state["output"] = action["content"]
                break
            
            result = self.execute(action)
            state["steps"].append({
                "action": action,
                "result": result,
                "duration_ms": (time.time() - step_start) * 1000,
            })
            self.step_count += 1
        
        duration = time.time() - start_time
        self.logger.info(f"Agent completed in {duration:.2f}s, "
                        f"{self.step_count} steps")
        
        # Cost tracking
        self.track_cost(self.step_count, duration)
        
        return state["output"]

import json

def simple_agent(user_input, tools, llm, max_steps=5):
    messages = [{"role": "user", "content": user_input}]
    
    for step in range(max_steps):
        # LLM decides next action
        response = llm.invoke(messages)
        messages.append(response)
        
        # Check if final answer
        if response.get("type") == "final":
            return response["content"]
        
        # Execute tool
        tool_call = json.loads(response["tool_call"])
        tool_name = tool_call["name"]
        tool_args = tool_call["arguments"]
        
        if tool_name in tools:
            result = tools[tool_name](**tool_args)
            messages.append({"role": "tool", "content": str(result), 
                            "tool_call_id": tool_call.get("id")})
        else:
            messages.append({"role": "tool", "content": f"Tool {tool_name} not found",
                            "tool_call_id": tool_call.get("id")})
    
    return "Max steps reached"

# Simulate a tool
def search(query):
    return f"Results for: {query}"

# Run
result = simple_agent(
    "Find the capital of France",
    {"search": search},
    lambda msgs: {"type": "final", "content": "Paris"},
)
print(result)