AI Agents: Build Autonomous Systems with LLMs

What are AI Agents?

AI agents are autonomous systems that use LLMs to reason, plan, and take actions to accomplish goals. Unlike simple chatbots, agents can use tools, access external data, and make decisions independently.

Agent vs Chatbot

Chatbot	AI Agent
Single response	Multi-step reasoning
Predefined flows	Dynamic decisions
Limited context	Tool access
Manual intervention	Autonomous action

Agent Architecture

User Goal
    ↓
┌─────────────────────────────────────┐
│           AI Agent Core             │
│  ┌─────────┐  ┌─────────────────┐  │
│  │ Planner │→ │ Reasoning (LLM) │  │
│  └─────────┘  └────────┬────────┘  │
│                        ↓           │
│  ┌─────────────────────────────┐   │
│  │      Tool Selection         │   │
│  └─────────────────────────────┘   │
└────────────────┬────────────────────┘
                 ↓
    ┌────────────┴────────────┐
    ↓            ↓            ↓
┌───────┐  ┌──────────┐  ┌────────┐
│ Tools │  │ Memory   │  │ APIs   │
└───────┘  └──────────┘  └────────┘

LangChain Agents

Basic Agent Setup

from langchain_openai import ChatOpenAI
from langchain.agents import AgentExecutor, create_react_agent
from langchain.tools import Tool
from langchain import hub

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

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

tools = [
    Tool(
        name="Search",
        func=search_web,
        description="Search the web for current information"
    ),
    Tool(
        name="Calculator",
        func=calculate,
        description="Calculate mathematical expressions"
    )
]

# Create agent
llm = ChatOpenAI(model="gpt-4", temperature=0)
prompt = hub.pull("hwchase17/react")

agent = create_react_agent(llm, tools, prompt)
agent_executor = AgentExecutor(
    agent=agent,
    tools=tools,
    verbose=True,
    max_iterations=5
)

# Run agent
result = agent_executor.invoke({
    "input": "What is the population of Tokyo and what is that divided by 1000?"
})
print(result["output"])

Custom Tools

from langchain.tools import BaseTool
from pydantic import BaseModel, Field
from typing import Optional, Type

class WeatherInput(BaseModel):
    location: str = Field(description="City name")
    unit: str = Field(default="celsius", description="Temperature unit")

class WeatherTool(BaseTool):
    name: str = "weather"
    description: str = "Get current weather for a location"
    args_schema: Type[BaseModel] = WeatherInput

    def _run(self, location: str, unit: str = "celsius") -> str:
        # Implement weather API call
        return f"Weather in {location}: 22°{unit[0].upper()}, Sunny"

    async def _arun(self, location: str, unit: str = "celsius") -> str:
        return self._run(location, unit)

class DatabaseQueryInput(BaseModel):
    query: str = Field(description="SQL query to execute")

class DatabaseTool(BaseTool):
    name: str = "database"
    description: str = "Execute SQL queries on the database"
    args_schema: Type[BaseModel] = DatabaseQueryInput

    def _run(self, query: str) -> str:
        # Implement database query
        if "SELECT" in query.upper():
            return "Query results: [...]"
        return "Query executed successfully"

tools = [WeatherTool(), DatabaseTool()]

Structured Output Agent

from langchain.output_parsers import PydanticOutputParser
from pydantic import BaseModel
from typing import List

class ResearchResult(BaseModel):
    topic: str
    summary: str
    key_points: List[str]
    sources: List[str]
    confidence: float

parser = PydanticOutputParser(pydantic_object=ResearchResult)

research_prompt = """
Research the following topic and provide structured output.

Topic: {topic}

{format_instructions}
"""

from langchain.prompts import PromptTemplate

prompt = PromptTemplate(
    template=research_prompt,
    input_variables=["topic"],
    partial_variables={"format_instructions": parser.get_format_instructions()}
)

chain = prompt | llm | parser

result = chain.invoke({"topic": "Quantum Computing Applications"})
print(f"Summary: {result.summary}")
print(f"Key Points: {result.key_points}")

ReAct Pattern

Implementation

from langchain.agents import AgentExecutor
from langchain.agents.format_scratchpad import format_log_to_str
from langchain.agents.output_parsers import ReActSingleInputOutputParser
from langchain.prompts import PromptTemplate

react_template = """Answer the following questions as best you can. You have access to the following tools:

{tools}

Use the following format:

Question: the input question you must answer
Thought: you should always think about what to do
Action: the action to take, should be one of [{tool_names}]
Action Input: the input to the action
Observation: the result of the action
... (this Thought/Action/Action Input/Observation can repeat N times)
Thought: I now know the final answer
Final Answer: the final answer to the original input question

Begin!

Question: {input}
Thought:{agent_scratchpad}"""

prompt = PromptTemplate.from_template(react_template)

# Create agent with ReAct
agent = (
    {
        "input": lambda x: x["input"],
        "agent_scratchpad": lambda x: format_log_to_str(x["intermediate_steps"]),
        "tools": lambda x: "\n".join([f"{t.name}: {t.description}" for t in tools]),
        "tool_names": lambda x: ", ".join([t.name for t in tools])
    }
    | prompt
    | llm
    | ReActSingleInputOutputParser()
)

agent_executor = AgentExecutor(agent=agent, tools=tools, verbose=True)

Multi-Agent Systems

Supervisor Pattern

from langchain_openai import ChatOpenAI
from langgraph.graph import StateGraph, END
from typing import TypedDict, Annotated, Sequence
import operator

class AgentState(TypedDict):
    messages: Annotated[Sequence[str], operator.add]
    next: str

def create_agent(llm, system_prompt: str):
    def agent_node(state: AgentState):
        messages = state["messages"]
        response = llm.invoke([
            {"role": "system", "content": system_prompt},
            {"role": "user", "content": messages[-1]}
        ])
        return {"messages": [response.content]}
    return agent_node

# Create specialized agents
researcher = create_agent(
    ChatOpenAI(model="gpt-4"),
    "You are a research specialist. Find and analyze information."
)

writer = create_agent(
    ChatOpenAI(model="gpt-4"),
    "You are a content writer. Create clear, engaging content."
)

reviewer = create_agent(
    ChatOpenAI(model="gpt-4"),
    "You are an editor. Review and improve content quality."
)

def supervisor(state: AgentState):
    """Route to next agent based on task progress."""
    messages = state["messages"]

    # Simple routing logic
    if len(messages) == 1:
        return {"next": "researcher"}
    elif len(messages) == 2:
        return {"next": "writer"}
    elif len(messages) == 3:
        return {"next": "reviewer"}
    else:
        return {"next": "end"}

# Build graph
workflow = StateGraph(AgentState)

workflow.add_node("supervisor", supervisor)
workflow.add_node("researcher", researcher)
workflow.add_node("writer", writer)
workflow.add_node("reviewer", reviewer)

workflow.set_entry_point("supervisor")

workflow.add_conditional_edges(
    "supervisor",
    lambda x: x["next"],
    {
        "researcher": "researcher",
        "writer": "writer",
        "reviewer": "reviewer",
        "end": END
    }
)

workflow.add_edge("researcher", "supervisor")
workflow.add_edge("writer", "supervisor")
workflow.add_edge("reviewer", "supervisor")

app = workflow.compile()

# Run
result = app.invoke({
    "messages": ["Write a blog post about AI agents"],
    "next": ""
})

AutoGen Multi-Agent

import autogen

config_list = [{"model": "gpt-4", "api_key": "your-key"}]

# Create agents
assistant = autogen.AssistantAgent(
    name="Assistant",
    llm_config={"config_list": config_list},
    system_message="You are a helpful AI assistant."
)

coder = autogen.AssistantAgent(
    name="Coder",
    llm_config={"config_list": config_list},
    system_message="You are a Python expert. Write clean, efficient code."
)

reviewer = autogen.AssistantAgent(
    name="Reviewer",
    llm_config={"config_list": config_list},
    system_message="You review code for bugs and improvements."
)

user_proxy = autogen.UserProxyAgent(
    name="User",
    human_input_mode="NEVER",
    max_consecutive_auto_reply=10,
    code_execution_config={"work_dir": "coding"}
)

# Group chat
group_chat = autogen.GroupChat(
    agents=[user_proxy, coder, reviewer],
    messages=[],
    max_round=10
)

manager = autogen.GroupChatManager(
    groupchat=group_chat,
    llm_config={"config_list": config_list}
)

# Start conversation
user_proxy.initiate_chat(
    manager,
    message="Create a Python function to analyze CSV files and generate reports."
)

Agent Memory

Conversation Memory

from langchain.memory import ConversationBufferMemory, ConversationSummaryMemory

# Buffer memory (stores all)
buffer_memory = ConversationBufferMemory(
    memory_key="chat_history",
    return_messages=True
)

# Summary memory (compresses old messages)
summary_memory = ConversationSummaryMemory(
    llm=llm,
    memory_key="chat_history",
    return_messages=True
)

# Use with agent
agent_executor = AgentExecutor(
    agent=agent,
    tools=tools,
    memory=buffer_memory,
    verbose=True
)

Vector Memory

from langchain_community.vectorstores import Chroma
from langchain_openai import OpenAIEmbeddings
from langchain.memory import VectorStoreRetrieverMemory

embeddings = OpenAIEmbeddings()
vectorstore = Chroma(embedding_function=embeddings)

retriever = vectorstore.as_retriever(search_kwargs={"k": 5})

vector_memory = VectorStoreRetrieverMemory(
    retriever=retriever,
    memory_key="relevant_history"
)

# Add to memory
vector_memory.save_context(
    {"input": "My favorite color is blue"},
    {"output": "I'll remember that your favorite color is blue."}
)

# Retrieve relevant memories
relevant = vector_memory.load_memory_variables(
    {"input": "What colors do I like?"}
)

Production Patterns

Error Handling

from tenacity import retry, stop_after_attempt, wait_exponential

class RobustAgent:
    def __init__(self, agent_executor):
        self.agent = agent_executor

    @retry(
        stop=stop_after_attempt(3),
        wait=wait_exponential(min=1, max=10)
    )
    def run(self, input_text: str) -> str:
        try:
            result = self.agent.invoke({"input": input_text})
            return result["output"]
        except Exception as e:
            print(f"Error: {e}, retrying...")
            raise

    def run_with_fallback(self, input_text: str) -> str:
        try:
            return self.run(input_text)
        except Exception as e:
            return f"I apologize, but I couldn't complete that task. Error: {e}"

Logging and Monitoring

import logging
from datetime import datetime

logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)

class MonitoredAgent:
    def __init__(self, agent_executor):
        self.agent = agent_executor
        self.metrics = {
            "total_runs": 0,
            "successful_runs": 0,
            "failed_runs": 0,
            "total_tokens": 0
        }

    def run(self, input_text: str) -> dict:
        start_time = datetime.now()
        self.metrics["total_runs"] += 1

        try:
            result = self.agent.invoke({"input": input_text})
            self.metrics["successful_runs"] += 1

            duration = (datetime.now() - start_time).total_seconds()

            logger.info(f"Agent completed in {duration:.2f}s")

            return {
                "output": result["output"],
                "duration": duration,
                "success": True
            }
        except Exception as e:
            self.metrics["failed_runs"] += 1
            logger.error(f"Agent failed: {e}")
            return {"output": None, "error": str(e), "success": False}

Summary

Pattern	Use Case
ReAct	General reasoning tasks
Plan-and-Execute	Complex multi-step tasks
Multi-Agent	Specialized collaboration
Tool-Use	External integrations

AI agents enable autonomous task completion with reasoning, planning, and tool use capabilities.