Agentic AI – Complete Tutorial

Introduction

Agentic AI refers to AI systems that can autonomously plan, reason, decide, and take actions to achieve complex goals — going beyond simple question-answering to executing multi-step tasks with minimal human intervention. Unlike traditional AI that responds to a single prompt, agentic systems can break down problems, use tools, maintain memory, collaborate with other agents, and adapt their strategy based on feedback.

Think of it this way: - Traditional AI: "What is the weather?" → "It's 72°F and sunny" - Agentic AI: "Plan my outdoor event for next weekend" → Checks weather forecasts, finds available venues, compares prices, sends invitations, sets up calendar reminders — all autonomously

Core Concepts

LLM as a Reasoning Engine

At the heart of every AI agent is a Large Language Model (LLM) acting as the "brain." Instead of generating text responses, the LLM is used to:

1. Understand the user's goal
2. Plan what steps are needed
3. Decide which tools to use
4. Evaluate results and adjust strategy

# Conceptual flow of an AI agent
def agent_loop(goal):
    plan = llm.plan(goal)              # Break goal into steps
    for step in plan:
        tool = llm.select_tool(step)   # Decide which tool to use
        result = tool.execute(step)    # Execute the action
        if llm.evaluate(result):       # Check if step succeeded
            continue
        else:
            plan = llm.replan(goal, result)  # Adapt strategy
    return final_result

The ReAct Pattern (Reasoning + Acting)

The most fundamental agentic pattern. The LLM alternates between thinking (reasoning) and doing (acting):

Thought: I need to find the current stock price of AAPL
Action: search_stock_price(ticker="AAPL")
Observation: AAPL is currently trading at $195.23
Thought: Now I need to compare it with last month's price
Action: get_historical_price(ticker="AAPL", days_ago=30)
Observation: 30 days ago AAPL was $187.45
Thought: I can now calculate the change and provide the answer
Final Answer: AAPL has risen $7.78 (4.15%) over the past 30 days.

Tool Use

Agents become powerful when they can interact with the outside world through tools:

from langchain.tools import tool

@tool
def search_web(query: str) -> str:
    """Search the web for current information."""
    results = google_search_api(query)
    return results

@tool
def execute_python(code: str) -> str:
    """Execute Python code and return the output."""
    return exec_sandbox(code)

@tool
def query_database(sql: str) -> str:
    """Run a SQL query against the company database."""
    return db.execute(sql)

@tool
def send_email(to: str, subject: str, body: str) -> str:
    """Send an email to the specified recipient."""
    return email_client.send(to, subject, body)

Memory Systems

Agents need memory to maintain context across interactions:

Short-Term Memory (Conversation Buffer)

from langchain.memory import ConversationBufferMemory

memory = ConversationBufferMemory()
# Stores recent conversation turns
# Limited by context window size

Long-Term Memory (Vector Store)

from langchain.vectorstores import Chroma
from langchain.embeddings import OpenAIEmbeddings

# Store knowledge that persists across sessions
vectorstore = Chroma(embedding_function=OpenAIEmbeddings())
vectorstore.add_documents(company_docs)

# Agent can retrieve relevant info anytime
relevant_docs = vectorstore.similarity_search("Q4 revenue")

Episodic Memory (Past Experiences)

# Store successful action sequences for future reference
memory_store = {
    "deploy_app": ["run tests", "build docker", "push to ECR", "update ECS"],
    "debug_error": ["check logs", "identify root cause", "apply fix", "verify"]
}

Planning Strategies

Sequential Planning — Steps executed one after another:

Goal: "Write a blog post about AI trends"
Plan:
  1. Research current AI trends (search web)
  2. Outline key points (reasoning)
  3. Write draft (text generation)
  4. Review and edit (self-critique)
  5. Format in markdown (tool use)

Hierarchical Planning — Break into sub-goals, each with their own plan:

Goal: "Launch new product feature"
├── Sub-goal 1: Design
│   ├── Research competitor features
│   ├── Create user stories
│   └── Design mockups
├── Sub-goal 2: Implement
│   ├── Write backend API
│   ├── Build frontend UI
│   └── Write tests
└── Sub-goal 3: Deploy
    ├── Set up CI/CD
    ├── Deploy to staging
    └── Monitor and release

Building Agents with LangChain

Basic Agent

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

# Define tools
@tool
def calculator(expression: str) -> str:
    """Calculate a mathematical expression."""
    return str(eval(expression))

@tool
def get_weather(city: str) -> str:
    """Get current weather for a city."""
    # In production, call a real weather API
    import requests
    resp = requests.get(f"https://wttr.in/{city}?format=3")
    return resp.text

# Set up the agent
llm = ChatOpenAI(model="gpt-4", temperature=0)
tools = [calculator, get_weather]
prompt = hub.pull("hwchase17/react")

agent = create_react_agent(llm, tools, prompt)
executor = AgentExecutor(agent=agent, tools=tools, verbose=True)

# Run the agent
result = executor.invoke({
    "input": "What's the weather in Tokyo? If the temperature is above 25°C, calculate 25 * 1.8 + 32 to convert to Fahrenheit"
})
print(result["output"])

Multi-Tool Research Agent

from langchain.agents import AgentExecutor, create_openai_tools_agent
from langchain_openai import ChatOpenAI
from langchain.tools import tool
from langchain_core.prompts import ChatPromptTemplate, MessagesPlaceholder

@tool
def search_arxiv(query: str) -> str:
    """Search academic papers on arXiv."""
    import arxiv
    results = arxiv.Search(query=query, max_results=3)
    papers = []
    for r in results.results():
        papers.append(f"Title: {r.title}\nSummary: {r.summary[:200]}\n")
    return "\n".join(papers)

@tool
def search_web(query: str) -> str:
    """Search the internet for current information."""
    from tavily import TavilyClient
    client = TavilyClient()
    results = client.search(query)
    return str(results["results"][:3])

@tool
def write_file(filename: str, content: str) -> str:
    """Write content to a file."""
    with open(filename, "w") as f:
        f.write(content)
    return f"Successfully wrote to {filename}"

# Create agent with system prompt
prompt = ChatPromptTemplate.from_messages([
    ("system", "You are a research assistant. Search for information, "
               "synthesize findings, and save reports to files."),
    ("human", "{input}"),
    MessagesPlaceholder("agent_scratchpad"),
])

llm = ChatOpenAI(model="gpt-4", temperature=0)
tools = [search_arxiv, search_web, write_file]
agent = create_openai_tools_agent(llm, tools, prompt)
executor = AgentExecutor(agent=agent, tools=tools, verbose=True)

result = executor.invoke({
    "input": "Research the latest advances in agentic AI systems. "
             "Find 3 academic papers and 3 industry articles, then "
             "write a summary report to 'agentic_ai_report.md'"
})

Building Agents with LangGraph

LangGraph provides more control over agent workflows using a graph-based approach:

Stateful Agent with Conditional Logic

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

class AgentState(TypedDict):
    messages: list
    next_step: str
    research_data: str
    draft: str
    final_output: str

llm = ChatOpenAI(model="gpt-4")

def research_node(state: AgentState) -> AgentState:
    """Gather information from various sources."""
    query = state["messages"][-1]
    # Simulate research
    research = llm.invoke(f"Research this topic thoroughly: {query}")
    return {"research_data": research.content, "next_step": "write"}

def write_node(state: AgentState) -> AgentState:
    """Write content based on research."""
    draft = llm.invoke(
        f"Based on this research:\n{state['research_data']}\n\n"
        f"Write a comprehensive article."
    )
    return {"draft": draft.content, "next_step": "review"}

def review_node(state: AgentState) -> AgentState:
    """Review and decide if content needs revision."""
    review = llm.invoke(
        f"Review this article for accuracy and completeness. "
        f"Respond with 'APPROVE' if good, or 'REVISE: [feedback]' if not.\n\n"
        f"{state['draft']}"
    )
    if "APPROVE" in review.content:
        return {"final_output": state["draft"], "next_step": "end"}
    else:
        return {"next_step": "write"}  # Loop back to rewrite

def route_next(state: AgentState) -> str:
    if state["next_step"] == "end":
        return END
    return state["next_step"]

# Build the graph
graph = StateGraph(AgentState)
graph.add_node("research", research_node)
graph.add_node("write", write_node)
graph.add_node("review", review_node)

graph.set_entry_point("research")
graph.add_edge("research", "write")
graph.add_edge("write", "review")
graph.add_conditional_edges("review", route_next)

app = graph.compile()

# Run
result = app.invoke({
    "messages": ["Write an article about quantum computing in 2025"],
    "next_step": "research",
    "research_data": "",
    "draft": "",
    "final_output": ""
})
print(result["final_output"])

Multi-Agent Systems with CrewAI

CrewAI enables teams of specialized agents collaborating on complex tasks:

from crewai import Agent, Task, Crew
from langchain_openai import ChatOpenAI

llm = ChatOpenAI(model="gpt-4")

# Define specialized agents
researcher = Agent(
    role="Senior Research Analyst",
    goal="Find and analyze the latest information on the given topic",
    backstory="You are an expert researcher with 20 years of experience "
              "in technology analysis. You excel at finding credible sources "
              "and extracting key insights.",
    llm=llm,
    verbose=True
)

writer = Agent(
    role="Technical Content Writer",
    goal="Write clear, engaging, and technically accurate content",
    backstory="You are a seasoned tech writer who can explain complex "
              "topics in simple terms while maintaining technical accuracy.",
    llm=llm,
    verbose=True
)

editor = Agent(
    role="Senior Editor",
    goal="Ensure content is polished, accurate, and publication-ready",
    backstory="You are a meticulous editor who catches errors, improves "
              "clarity, and ensures consistent tone and style.",
    llm=llm,
    verbose=True
)

# Define tasks
research_task = Task(
    description="Research the current state of AI agents in enterprise "
                "applications. Find real-world examples, market data, "
                "and expert opinions. Focus on 2024-2025 developments.",
    agent=researcher,
    expected_output="A detailed research brief with sources and key findings"
)

writing_task = Task(
    description="Using the research findings, write a 1500-word article "
                "about AI agents in enterprise. Include an introduction, "
                "3-4 main sections with examples, and a conclusion.",
    agent=writer,
    expected_output="A well-structured 1500-word article in markdown format"
)

editing_task = Task(
    description="Review and polish the article. Check for factual accuracy, "
                "improve readability, fix any grammatical issues, and ensure "
                "the article flows well from introduction to conclusion.",
    agent=editor,
    expected_output="A final, publication-ready article"
)

# Create and run the crew
crew = Crew(
    agents=[researcher, writer, editor],
    tasks=[research_task, writing_task, editing_task],
    verbose=True
)

result = crew.kickoff()
print(result)

Real-World Use Cases

1. Customer Support Agent

An agent that handles customer inquiries by searching knowledge bases, checking order status, and escalating when needed:

from langchain.agents import AgentExecutor, create_openai_tools_agent
from langchain_openai import ChatOpenAI
from langchain.tools import tool

@tool
def search_knowledge_base(query: str) -> str:
    """Search the company FAQ and documentation."""
    # Vector similarity search against company docs
    results = vectorstore.similarity_search(query, k=3)
    return "\n".join([doc.page_content for doc in results])

@tool
def check_order_status(order_id: str) -> str:
    """Check the status of a customer order."""
    order = db.query(f"SELECT * FROM orders WHERE id = '{order_id}'")
    return f"Order {order_id}: {order['status']}, ETA: {order['delivery_date']}"

@tool
def create_support_ticket(issue: str, priority: str) -> str:
    """Create a support ticket for issues requiring human review."""
    ticket_id = ticketing_system.create(issue=issue, priority=priority)
    return f"Ticket {ticket_id} created. A human agent will follow up within 24 hours."

@tool
def process_refund(order_id: str, reason: str) -> str:
    """Process a refund for an order (max $100 without approval)."""
    order = db.query(f"SELECT amount FROM orders WHERE id = '{order_id}'")
    if order["amount"] <= 100:
        payment_system.refund(order_id)
        return f"Refund of ${order['amount']} processed for order {order_id}"
    return "Refund exceeds $100 limit. Escalating to supervisor."

# System prompt defines agent behavior
system_prompt = """You are a helpful customer support agent for TechStore.
Rules:
- Always greet the customer warmly
- Search the knowledge base before saying "I don't know"
- Process refunds under $100 automatically
- Escalate complex issues by creating tickets
- Never share internal system details with customers
- Always confirm actions before executing them"""

2. Data Analysis Agent

An agent that can query databases, create visualizations, and generate reports:

@tool
def query_sql(question: str) -> str:
    """Convert a natural language question to SQL and execute it."""
    # LLM generates SQL from the question
    sql = llm.invoke(
        f"Given tables: sales(date, product, revenue, region), "
        f"customers(id, name, segment, lifetime_value)\n"
        f"Write SQL for: {question}"
    ).content
    result = db.execute(sql)
    return f"Query: {sql}\nResult: {result}"

@tool
def create_chart(data: str, chart_type: str, title: str) -> str:
    """Create a chart from data and save as image."""
    import matplotlib.pyplot as plt
    import json

    parsed = json.loads(data)
    fig, ax = plt.subplots()
    if chart_type == "bar":
        ax.bar(parsed["labels"], parsed["values"])
    elif chart_type == "line":
        ax.plot(parsed["labels"], parsed["values"])
    ax.set_title(title)
    filename = f"charts/{title.replace(' ', '_')}.png"
    plt.savefig(filename)
    return f"Chart saved to {filename}"

@tool
def generate_report(findings: str, format: str) -> str:
    """Generate a formatted report from analysis findings."""
    report = llm.invoke(
        f"Create a professional {format} report from these findings:\n{findings}"
    ).content
    with open(f"reports/analysis_report.{format}", "w") as f:
        f.write(report)
    return "Report generated successfully"

# Usage: "Analyze Q4 sales trends by region, create a bar chart,
#          and generate a PDF report with recommendations"

3. DevOps Automation Agent

An agent that monitors systems, diagnoses issues, and applies fixes:

@tool
def check_service_health(service_name: str) -> str:
    """Check health status of a microservice."""
    response = requests.get(f"http://{service_name}/health")
    metrics = cloudwatch.get_metrics(service_name, period=300)
    return f"Status: {response.status_code}, CPU: {metrics['cpu']}%, Memory: {metrics['memory']}%"

@tool
def get_logs(service_name: str, minutes: int) -> str:
    """Retrieve recent logs from CloudWatch."""
    logs = cloudwatch_logs.filter(
        log_group=f"/ecs/{service_name}",
        start_time=datetime.now() - timedelta(minutes=minutes)
    )
    return "\n".join([log["message"] for log in logs[:50]])

@tool
def scale_service(service_name: str, desired_count: int) -> str:
    """Scale an ECS service to the desired number of tasks."""
    ecs.update_service(
        cluster="production",
        service=service_name,
        desiredCount=desired_count
    )
    return f"Scaled {service_name} to {desired_count} tasks"

@tool
def restart_service(service_name: str) -> str:
    """Force restart a service by stopping all tasks."""
    ecs.update_service(
        cluster="production",
        service=service_name,
        forceNewDeployment=True
    )
    return f"Restarting {service_name}..."

@tool
def create_incident(title: str, severity: str, details: str) -> str:
    """Create a PagerDuty incident for the on-call team."""
    incident = pagerduty.create_incident(
        title=title, severity=severity, body=details
    )
    return f"Incident created: {incident['id']}"

# Agent can autonomously: detect high CPU → check logs → identify memory leak
# → scale service → create incident → notify team

4. Code Review Agent

@tool
def get_pull_request_diff(pr_number: int) -> str:
    """Fetch the diff of a pull request from GitHub."""
    pr = github.get_repo("org/repo").get_pull(pr_number)
    return pr.get_files()

@tool
def check_code_style(code: str, language: str) -> str:
    """Run linting and style checks on code."""
    if language == "python":
        result = subprocess.run(["ruff", "check", "--stdin"], input=code)
    return result.stdout

@tool
def run_tests(test_path: str) -> str:
    """Run unit tests and return results."""
    result = subprocess.run(["pytest", test_path, "--tb=short"])
    return result.stdout

@tool
def post_review_comment(pr_number: int, comment: str, file: str, line: int) -> str:
    """Post a review comment on a specific line of a PR."""
    pr = github.get_repo("org/repo").get_pull(pr_number)
    pr.create_review_comment(body=comment, path=file, line=line)
    return "Comment posted"

# Agent reviews PRs: checks style → identifies bugs → suggests improvements
# → verifies tests pass → posts inline comments

5. Personal Research Assistant

@tool
def search_papers(query: str, num_results: int = 5) -> str:
    """Search academic papers on Google Scholar/arXiv."""
    papers = scholarly.search(query, limit=num_results)
    return format_papers(papers)

@tool
def summarize_pdf(url: str) -> str:
    """Download and summarize a PDF document."""
    content = extract_pdf_text(url)
    summary = llm.invoke(f"Summarize this paper:\n{content[:5000]}")
    return summary.content

@tool
def save_notes(topic: str, notes: str) -> str:
    """Save research notes to the knowledge base."""
    vectorstore.add_texts([notes], metadatas=[{"topic": topic}])
    return f"Notes saved under topic: {topic}"

@tool
def compare_papers(paper_ids: list) -> str:
    """Compare methodology and findings across multiple papers."""
    papers = [get_paper_content(pid) for pid in paper_ids]
    comparison = llm.invoke(
        f"Compare these papers on methodology, findings, and limitations:\n"
        + "\n---\n".join(papers)
    )
    return comparison.content

Agent Design Patterns

Pattern 1: Router Agent

Routes incoming requests to specialized sub-agents:

def router_agent(query: str) -> str:
    classification = llm.invoke(
        f"Classify this query into one of: [technical, billing, general]\n"
        f"Query: {query}\nCategory:"
    ).content.strip()

    if classification == "technical":
        return technical_agent.invoke(query)
    elif classification == "billing":
        return billing_agent.invoke(query)
    else:
        return general_agent.invoke(query)

Pattern 2: Critic/Refiner Loop

Agent generates output, then critiques and refines it:

def generate_with_critique(task: str, max_iterations: int = 3) -> str:
    draft = llm.invoke(f"Complete this task: {task}").content

    for i in range(max_iterations):
        critique = llm.invoke(
            f"Critique this output. List specific issues or say 'APPROVED':\n{draft}"
        ).content

        if "APPROVED" in critique:
            return draft

        draft = llm.invoke(
            f"Improve this based on feedback:\nOriginal: {draft}\nFeedback: {critique}"
        ).content

    return draft

Pattern 3: Parallel Execution

Multiple agents work on different sub-tasks simultaneously:

import asyncio

async def parallel_research(topic: str):
    tasks = [
        asyncio.create_task(web_researcher.invoke(topic)),
        asyncio.create_task(academic_researcher.invoke(topic)),
        asyncio.create_task(news_researcher.invoke(topic)),
    ]
    results = await asyncio.gather(*tasks)
    # Synthesizer combines all findings
    synthesis = synthesizer_agent.invoke(
        f"Combine these findings into a coherent report:\n{results}"
    )
    return synthesis

Pattern 4: Human-in-the-Loop

Agent pauses for human approval on critical actions:

def agent_with_approval(task: str):
    plan = llm.invoke(f"Create a plan for: {task}").content

    for step in parse_plan(plan):
        if step.is_critical:  # e.g., sending money, deleting data
            print(f"\n⚠️  Agent wants to: {step.description}")
            approval = input("Approve? (yes/no): ")
            if approval != "yes":
                continue
        step.execute()

Frameworks Comparison

Framework	Best For	Complexity	Key Feature
LangChain	General-purpose agents	Medium	Large ecosystem, many integrations
LangGraph	Complex workflows with state	High	Graph-based flow control, cycles
CrewAI	Multi-agent collaboration	Medium	Role-based agents, delegation
AutoGen (Microsoft)	Conversational multi-agent	Medium	Agent-to-agent conversations
OpenAI Assistants	Quick prototyping	Low	Built-in tools, managed threads
Amazon Bedrock Agents	AWS-native applications	Medium	Managed, integrated with AWS

Best Practices

Safety & Guardrails

# Always implement safety checks
class SafeAgent:
    def __init__(self, agent, max_iterations=10, max_cost=1.0):
        self.agent = agent
        self.max_iterations = max_iterations
        self.max_cost = max_cost
        self.iteration_count = 0
        self.total_cost = 0

    def invoke(self, input):
        self.iteration_count += 1
        if self.iteration_count > self.max_iterations:
            raise Exception("Max iterations reached - possible infinite loop")
        if self.total_cost > self.max_cost:
            raise Exception("Cost limit exceeded")
        return self.agent.invoke(input)

Key Principles

1. Start simple — Begin with a single agent and few tools; add complexity gradually
2. Limit scope — Define clear boundaries for what the agent can and cannot do
3. Implement guardrails — Set max iterations, cost limits, and action restrictions
4. Log everything — Record all agent decisions, tool calls, and outputs for debugging
5. Human oversight — Keep humans in the loop for critical actions (payments, deletions, external communications)
6. Test adversarially — Try to break your agent with edge cases and malicious inputs
7. Use structured outputs — Force agents to respond in predictable formats (JSON, function calls)
8. Handle failures gracefully — Agents will make mistakes; design for recovery

Getting Started

Prerequisites

pip install langchain langchain-openai langgraph crewai tavily-py

Environment Setup

export OPENAI_API_KEY="your-key-here"
export TAVILY_API_KEY="your-key-here"  # For web search

Your First Agent in 20 Lines

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

@tool
def multiply(a: int, b: int) -> int:
    """Multiply two numbers together."""
    return a * b

llm = ChatOpenAI(model="gpt-4", temperature=0)
prompt = hub.pull("hwchase17/react")
agent = create_react_agent(llm, [multiply], prompt)
executor = AgentExecutor(agent=agent, tools=[multiply], verbose=True)

result = executor.invoke({"input": "What is 23 multiplied by 47?"})
print(result["output"])  # 1081

What's Next

• Explore Amazon Bedrock Agents for managed agentic workflows on AWS
• Learn LangGraph for complex multi-step workflows with state management
• Build multi-agent systems with CrewAI for collaborative AI teams
• Implement RAG + Agents for knowledge-grounded autonomous systems