What Is an AI Agent?
An AI agent is an autonomous system that can perceive its environment, make decisions, and take actions to achieve specific goals. Unlike a simple chatbot that responds to individual prompts, an AI agent can:
- Reason about complex problems by breaking them into steps
- Use tools to interact with external systems (APIs, databases, file systems)
- Remember context across multiple interactions
- Plan multi-step strategies to accomplish tasks
- Adapt its behavior based on feedback and results
Think of an AI agent as a digital assistant that doesn't just answer questions. it does things. It can search the web, read files, call APIs, send emails, and chain these actions together to complete complex workflows.
Agents vs. Chatbots vs. Workflows
| Feature | Chatbot | Workflow | AI Agent | |---------|---------|----------|----------| | Autonomy | None | Fixed paths | Dynamic decisions | | Tool use | No | Pre-defined | On-the-fly selection | | Memory | Per-session | None | Persistent | | Planning | No | Hardcoded | Self-directed | | Adaptation | No | No | Yes |
Why Build Your Own AI Agent?
Building your own AI agent gives you complete control over:
- Customization: Tailor the agent to your specific use case. customer support, data analysis, content generation, research automation, and more.
- Privacy: Keep sensitive data within your own infrastructure instead of sending it to third-party services.
- Cost efficiency: Avoid per-seat SaaS pricing and pay only for the API calls you make.
- Integration: Connect to your existing tools, databases, and APIs exactly how you need.
- Learning: Understanding agent architecture is becoming an essential skill as AI adoption accelerates across industries.
Prerequisites & Tools
What You Need Before Starting
- Python 3.9+ installed on your system
- Node.js 18+ (for some tooling and deployment)
- A code editor (VS Code recommended)
- An OpenAI API key (sign up at [platform.openai.com](https://platform.openai.com))
- Basic knowledge of Python and JavaScript
Key Libraries and Frameworks
| Tool | Purpose | |------|---------| | OpenAI Python SDK | Interface with GPT models | | LangChain | Agent framework and tool orchestration | | LangGraph | Advanced agent workflows with state management | | python-dotenv | Environment variable management | | FastAPI | Building the agent's API server | | Render | Cloud deployment platform |
Understanding AI Agent Architecture
Before writing any code, it's essential to understand the core components that make up an AI agent.
The Core Loop
Every AI agent follows a fundamental reasoning cycle:
`` ┌─────────────────────────────────────────────┐ │ USER INPUT │ │ "Research competitors" │ └──────────────────┬──────────────────────────┘ ▼ ┌─────────────────────────────────────────────┐ │ LLM REASONING │ │ "I need to search the web first, then │ │ compile the results into a report." │ └──────────────────┬──────────────────────────┘ ▼ ┌─────────────────────────────────────────────┐ │ TOOL SELECTION │ │ "I'll use the web_search tool with │ │ query 'top competitors in AI agents'" │ └──────────────────┬──────────────────────────┘ ▼ ┌─────────────────────────────────────────────┐ │ TOOL EXECUTION │ │ web_search("top competitors in AI agents") │ │ → Returns list of search results │ └──────────────────┬──────────────────────────┘ ▼ ┌─────────────────────────────────────────────┐ │ OBSERVATION │ │ "I have the search results. Now I need │ │ to summarize them into a report." │ └──────────────────┬──────────────────────────┘ ▼ ┌─────────────────────────────────────────────┐ │ FINAL RESPONSE │ │ "Here are the top competitors in the │ │ AI agent space, with analysis..." │ └─────────────────────────────────────────────┘ `
Key Components
- LLM (Large Language Model): The "brain" that reasons, plans, and generates responses. In this tutorial, we use OpenAI's GPT models.
- Tools: Functions the agent can call to interact with the outside world. web search, file reading, API calls, database queries, and more.
- Memory: Stores context from previous interactions so the agent can maintain continuity across conversations.
- Agent Loop: The orchestration logic that decides when to think, when to use a tool, and when to respond to the user.
- Prompt Templates: Structured instructions that tell the agent how to behave, what tools are available, and how to format responses.
Setting Up Your Development Environment
Step 1: Create the Project
`bash
mkdir ai-agent-project
cd ai-agent-project
python -m venv venv
# On Windows venv\Scripts\activate
# On macOS/Linux source venv/bin/activate
`
Step 2: Install Dependencies
Create a
requirements.txt file:
`text
openai==1.35.0
langchain==0.2.10
langchain-openai==0.1.17
langchain-community==0.2.9
langgraph==0.1.1
python-dotenv==1.0.0
fastapi==0.111.0
uvicorn==0.30.1
pydantic==2.7.4
requests==2.32.3
`
Install the packages:
`bash
pip install -r requirements.txt
`
Step 3: Set Up Environment Variables
Create a
.env file in your project root:
`env
OPENAI_API_KEY=sk-your-api-key-here
MODEL_NAME=gpt-4o
TEMPERATURE=0.7
`
Create a
config.py to load these settings:
`python
import os
from dotenv import load_dotenv
load_dotenv()
OPENAI_API_KEY = os.getenv("OPENAI_API_KEY") MODEL_NAME = os.getenv("MODEL_NAME", "gpt-4o") TEMPERATURE = float(os.getenv("TEMPERATURE", "0.7"))
if not OPENAI_API_KEY: raise ValueError("OPENAI_API_KEY is not set in .env file")
`
OpenAI API Integration
The OpenAI API is the foundation of our agent. Let's start by making basic API calls, then build up to agent functionality.
Basic API Call
`python
import openai
from config import OPENAI_API_KEY, MODEL_NAME
client = openai.OpenAI(api_key=OPENAI_API_KEY)
response = client.chat.completions.create( model=MODEL_NAME, messages=[ {"role": "system", "content": "You are a helpful AI assistant."}, {"role": "user", "content": "Explain what an AI agent is in 3 sentences."} ], temperature=0.7, max_tokens=500 )
print(response.choices[0].message.content)
`
Understanding the Response Object
`python
# The response contains useful metadata
print(f"Model used: {response.model}")
print(f"Tokens used: {response.usage.total_tokens}")
print(f"Prompt tokens: {response.usage.prompt_tokens}")
print(f"Completion tokens: {response.usage.completion_tokens}")
print(f"Finish reason: {response.choices[0].finish_reason}")
`
Streaming Responses
For a better user experience, stream the response token by token:
`python
def stream_response(messages):
stream = client.chat.completions.create(
model=MODEL_NAME,
messages=messages,
stream=True
)
full_response = ""
for chunk in stream:
if chunk.choices[0].delta.content is not None:
content = chunk.choices[0].delta.content
full_response += content
print(content, end="", flush=True)
return full_response
messages = [ {"role": "system", "content": "You are a helpful AI assistant."}, {"role": "user", "content": "Write a haiku about programming."} ]
stream_response(messages)
`
Structured Output with Function Calling
OpenAI's function calling feature lets the model request specific tool invocations:
`python
import json
# Define available tools tools = [ { "type": "function", "function": { "name": "get_weather", "description": "Get the current weather for a location", "parameters": { "type": "object", "properties": { "location": { "type": "string", "description": "The city name, e.g. 'New York'" }, "unit": { "type": "string", "enum": ["celsius", "fahrenheit"], "description": "Temperature unit" } }, "required": ["location"] } } }, { "type": "function", "function": { "name": "search_web", "description": "Search the web for information", "parameters": { "type": "object", "properties": { "query": { "type": "string", "description": "The search query" }, "num_results": { "type": "integer", "description": "Number of results to return" } }, "required": ["query"] } } } ]
# Make a request with tools response = client.chat.completions.create( model=MODEL_NAME, messages=[ {"role": "user", "content": "What's the weather in Tokyo?"} ], tools=tools, tool_choice="auto" )
# Check if the model wants to call a tool message = response.choices[0].message if message.tool_calls: for tool_call in message.tool_calls: print(f"Tool: {tool_call.function.name}") print(f"Arguments: {tool_call.function.arguments}")
`
Introduction to LangChain
LangChain is the most popular framework for building AI applications. It provides abstractions for working with LLMs, managing prompts, chaining operations, and. most importantly. building agents.
Why LangChain?
- Unified interface: Swap between different LLM providers with minimal code changes
- Pre-built components: Prompt templates, output parsers, document loaders, and more
- Agent frameworks: Built-in support for ReAct, tool-calling agents, and multi-agent systems
- Community tools: Hundreds of integrations with external services
LangChain Core Concepts
#### 1. Language Models
`python
from langchain_openai import ChatOpenAI
llm = ChatOpenAI( model="gpt-4o", temperature=0.7, api_key=OPENAI_API_KEY )
# Simple invocation response = llm.invoke("What is the capital of France?") print(response.content)
`
#### 2. Prompt Templates
`python
from langchain_core.prompts import ChatPromptTemplate
prompt = ChatPromptTemplate.from_messages([ ("system", "You are a helpful assistant that translates {input_language} to {output_language}."), ("human", "{text}") ])
# Format the prompt formatted = prompt.format_messages( input_language="English", output_language="Spanish", text="Hello, how are you?" )
response = llm.invoke(formatted) print(response.content)
`
#### 3. Output Parsers
`python
from langchain_core.output_parsers import JsonOutputParser
from pydantic import BaseModel, Field
class MovieReview(BaseModel): title: str = Field(description="The movie title") rating: int = Field(description="Rating from 1 to 10") summary: str = Field(description="Brief summary of the review") pros: list[str] = Field(description="List of positive aspects") cons: list[str] = Field(description="List of negative aspects")
parser = JsonOutputParser(pydantic_object=MovieReview)
prompt = ChatPromptTemplate.from_messages([ ("system", "You are a movie critic. Review the movie and format as JSON.\n{format_instructions}"), ("human", "Review: {movie_title}") ])
chain = prompt | llm | parser
result = chain.invoke({ "movie_title": "Inception", "format_instructions": parser.get_format_instructions() })
print(result) # {'title': 'Inception', 'rating': 9, 'summary': '...', 'pros': [...], 'cons': [...]}
`
#### 4. Chains
Chains let you compose multiple operations together:
`python
from langchain_core.prompts import PromptTemplate
from langchain_core.runnables import RunnablePassthrough
# Chain: generate a topic → write a blog post → create a title topic_prompt = PromptTemplate.from_template("Generate a blog post topic about {subject}") blog_prompt = PromptTemplate.from_template("Write a short blog post about this topic: {topic}") title_prompt = PromptTemplate.from_template("Create a catchy title for this blog post: {blog_post}")
# Build the chain from langchain_core.output_parsers import StrOutputParser
topic_chain = topic_prompt | llm | StrOutputParser() blog_chain = blog_prompt | llm | StrOutputParser() title_chain = title_prompt | llm | StrOutputParser()
# Compose into a full pipeline full_chain = ( {"topic": topic_chain} | RunnablePassthrough.assign(blog_post=blog_chain) | RunnablePassthrough.assign(title=title_chain) )
result = full_chain.invoke({"subject": "artificial intelligence"}) print(f"Title: {result['title']}\n\n{result['blog_post']}")
`
Building the Agent Logic
Now let's build the core agent. We'll start with a simple ReAct (Reasoning + Acting) agent, then enhance it.
The ReAct Pattern
The ReAct pattern is the foundation of most AI agents:
- Thought: The LLM reasons about what to do
- Action: It selects and calls a tool
- Observation: It receives the tool's result
- Repeat: It continues until the task is complete
Building a Basic Agent from Scratch
`python
import json
import re
from openai import OpenAI
client = OpenAI(api_key=OPENAI_API_KEY)
# Define tools as functions def calculator(expression: str) -> str: """Evaluate a mathematical expression.""" try: result = eval(expression, {"__builtins__": {}}, { "abs": abs, "round": round, "min": min, "max": max, "sum": sum, "len": len }) return str(result) except Exception as e: return f"Error: {e}"
def web_search(query: str) -> str: """Simulated web search (replace with real API in production).""" # In production, use SerpAPI, Tavily, or similar return f"Search results for '{query}': [Result 1] [Result 2] [Result 3]"
# Tool definitions for OpenAI tools = [ { "type": "function", "function": { "name": "calculator", "description": "Evaluate mathematical expressions. Input should be a valid math expression like '2 + 2' or 'math.sqrt(16)'.", "parameters": { "type": "object", "properties": { "expression": {"type": "string", "description": "The math expression to evaluate"} }, "required": ["expression"] } } }, { "type": "function", "function": { "name": "web_search", "description": "Search the web for current information. Use for facts, news, weather, and real-time data.", "parameters": { "type": "object", "properties": { "query": {"type": "string", "description": "The search query"} }, "required": ["query"] } } } ]
# Tool dispatch table tool_functions = { "calculator": calculator, "web_search": web_search }
def run_agent(user_message: str, max_iterations: int = 10): """Run the ReAct agent loop.""" system_prompt = """You are a helpful AI assistant with access to tools. When you need to use a tool, respond with a JSON object in this format: {"tool": "tool_name", "input": {"param": "value"}}
When you have the final answer, respond with plain text (no JSON).
Always think step by step about what you need to do.""" messages = [ {"role": "system", "content": system_prompt}, {"role": "user", "content": user_message} ] for i in range(max_iterations): # Get LLM response response = client.chat.completions.create( model=MODEL_NAME, messages=messages, tools=tools, tool_choice="auto" ) message = response.choices[0].message messages.append(message) # Check if the model wants to call a tool if message.tool_calls: print(f"\n🔧 [Step {i+1}] Tool call: {message.tool_calls[0].function.name}") for tool_call in message.tool_calls: function_name = tool_call.function.name function_args = json.loads(tool_call.function.arguments) # Execute the tool if function_name in tool_functions: result = tool_functions[function_name](function_args) print(f" Result: {result[:100]}...") else: result = f"Error: Unknown tool '{function_name}'" # Add tool result to messages messages.append({ "role": "tool", "tool_call_id": tool_call.id, "content": result }) else: # No tool calls. this is the final answer print(f"\n✅ Final Answer:\n{message.content}") return message.content return "Max iterations reached without a final answer."
# Test the agent run_agent("What is 15% of 240, and who won the 2024 US Open?")
`
Building with LangChain's Agent Framework
LangChain provides a much more solid agent implementation:
`python
from langchain_openai import ChatOpenAI
from langchain_core.prompts import ChatPromptTemplate, MessagesPlaceholder
from langchain.agents import create_tool_calling_agent, AgentExecutor
from langchain_core.tools import Tool
# Define tools using LangChain's Tool wrapper tools = [ Tool( name="calculator", func=calculator, description="Evaluate mathematical expressions. Input should be a valid math expression." ), Tool( name="web_search", func=web_search, description="Search the web for current information. Use for facts, news, weather, and real-time data." ) ]
# Create the prompt template prompt = ChatPromptTemplate.from_messages([ ("system", """You are a helpful AI assistant. Use the available tools to answer questions. Think step by step:
- What information do I need?
- Which tool can provide that information?
- What parameters does the tool need?
- After getting the result, do I need to do more, or can I answer?"""),
MessagesPlaceholder(variable_name="chat_history", optional=True), ("human", "{input}"), MessagesPlaceholder(variable_name="agent_scratchpad") ])
# Create the LLM llm = ChatOpenAI(model="gpt-4o", temperature=0)
# Create the agent agent = create_tool_calling_agent(llm, tools, prompt)
# Create the executor agent_executor = AgentExecutor( agent=agent, tools=tools, verbose=True, max_iterations=10, handle_parsing_errors=True )
# Run the agent result = agent_executor.invoke({ "input": "What is 15% of 240, and what is the square root of that?" })
print(f"\nFinal Answer: {result['output']}")
`
Adding Memory and Context
A key feature of agents is the ability to remember previous interactions. Let's add memory to our agent.
Conversation Memory with LangChain
`python
from langchain.memory import ConversationBufferMemory
from langchain_core.messages import HumanMessage, AIMessage
# Create memory that stores the full conversation memory = ConversationBufferMemory( memory_key="chat_history", return_messages=True )
# Create agent with memory agent_executor = AgentExecutor( agent=agent, tools=tools, memory=memory, verbose=True, max_iterations=10 )
# First interaction result1 = agent_executor.invoke({ "input": "My name is Alice and I live in New York." }) print(f"Agent: {result1['output']}")
# Second interaction. the agent remembers! result2 = agent_executor.invoke({ "input": "What city do I live in?" }) print(f"Agent: {result2['output']}") # The agent will answer "New York" because it remembers the previous message
`
Windowed Memory (for long conversations)
For production agents, you typically don't want to send the entire conversation history (it's expensive and can exceed context limits). Use a windowed approach:
`python
from langchain.memory import ConversationWindowBufferMemory
# Only keep the last 10 messages memory = ConversationWindowBufferMemory( memory_key="chat_history", k=10, # Keep last 10 messages return_messages=True )
`
Persistent Memory with a Database
For truly persistent memory across sessions:
`python
from langchain.memory import ConversationSummaryBufferMemory
# This summarizes old messages when the buffer is full memory = ConversationSummaryBufferMemory( llm=llm, memory_key="chat_history", max_token_limit=2000, # When exceeded, summarize older messages return_messages=True )
`
Working with Tools and Functions
Tools are what make agents powerful. Let's build a thorough set of tools.
Custom Tool: File Operations
`python
from langchain_core.tools import tool
from langchain_core.pydantic_v1 import BaseModel, Field
class FileReadInput(BaseModel): filepath: str = Field(description="Path to the file to read")
class FileWriteInput(BaseModel): filepath: str = Field(description="Path to the file to write") content: str = Field(description="Content to write to the file")
@tool(args_schema=FileReadInput) def read_file(filepath: str) -> str: """Read the contents of a file.""" try: with open(filepath, 'r', encoding='utf-8') as f: return f.read() except FileNotFoundError: return f"Error: File '{filepath}' not found" except Exception as e: return f"Error: {e}"
@tool(args_schema=FileWriteInput) def write_file(filepath: str, content: str) -> str: """Write content to a file.""" try: with open(filepath, 'w', encoding='utf-8') as f: f.write(content) return f"Successfully wrote {len(content)} characters to {filepath}" except Exception as e: return f"Error: {e}"
`
Custom Tool: API Integration
`python
import requests
@tool def get_weather(city: str) -> str: """Get the current weather for a city using OpenWeatherMap API.""" api_key = os.getenv("OPENWEATHER_API_KEY") url = f"http://api.openweathermap.org/data/2.5/weather?q={city}&appid={api_key}&units=metric" try: response = requests.get(url) data = response.json() if response.status_code == 200: temp = data["main"]["temp"] description = data["weather"][0]["description"] humidity = data["main"]["humidity"] return f"Weather in {city}: {description}, {temp}°C, humidity {humidity}%" else: return f"Error: {data.get('message', 'Unknown error')}" except Exception as e: return f"Error fetching weather: {e}"
@tool def fetch_url(url: str) -> str: """Fetch and extract text content from a URL.""" try: response = requests.get(url, timeout=10) response.raise_for_status() # Simple HTML text extraction (use BeautifulSoup for production) from html.parser import HTMLParser class HTMLTextExtractor(HTMLParser): def __init__(self): super().__init__() self.result = [] def handle_data(self, data): self.result.append(data) def get_text(self): return ' '.join(self.result) extractor = HTMLTextExtractor() extractor.feed(response.text) text = extractor.get_text() # Truncate to reasonable length return text[:5000] if len(text) > 5000 else text except Exception as e: return f"Error fetching URL: {e}"
`
Custom Tool: Database Query
`python
import sqlite3
@tool def query_database(query: str) -> str: """Execute a SQL query on the local SQLite database. Use SELECT queries only.""" try: conn = sqlite3.connect("agent_data.db") cursor = conn.cursor() cursor.execute(query) if query.strip().upper().startswith("SELECT"): rows = cursor.fetchall() columns = [desc[0] for desc in cursor.description] # Format as readable output result = ", ".join(columns) + "\n" for row in rows[:50]: # Limit to 50 rows result += ", ".join(str(cell) for cell in row) + "\n" if len(rows) > 50: result += f"... ({len(rows) - 50} more rows)" conn.close() return result else: conn.commit() conn.close() return f"Query executed successfully. {cursor.rowcount} rows affected." except Exception as e: return f"Database error: {e}"
`
Assembling the Full Tool Set
`python
all_tools = [
calculator,
web_search,
get_weather,
fetch_url,
read_file,
write_file,
query_database
]
# Create the agent with all tools prompt = ChatPromptTemplate.from_messages([ ("system", """You are a powerful AI assistant with access to multiple tools.
Available tools:
- calculator: For mathematical calculations
- web_search: For searching the internet
- get_weather: For weather information
- fetch_url: For reading web pages
- read_file: For reading local files
- write_file: For writing to local files
- query_database: For querying the SQLite database
Think step by step. Break complex tasks into smaller steps. Always verify your work before giving a final answer."""), MessagesPlaceholder(variable_name="chat_history", optional=True), ("human", "{input}"), MessagesPlaceholder(variable_name="agent_scratchpad") ])
agent = create_tool_calling_agent(llm, all_tools, prompt) agent_executor = AgentExecutor( agent=agent, tools=all_tools, memory=ConversationBufferMemory(memory_key="chat_history", return_messages=True), verbose=True, max_iterations=15, handle_parsing_errors=True )
`
Real-World Examples
Let's build three complete, real-world agent applications.
Example 1: Research Assistant Agent
`python
"""
Research Assistant Agent
Searches the web, reads articles, and compiles research reports.
"""
from langchain_openai import ChatOpenAI from langchain.agents import create_tool_calling_agent, AgentExecutor from langchain_core.tools import Tool from langchain_core.prompts import ChatPromptTemplate, MessagesPlaceholder from langchain.memory import ConversationBufferMemory import requests from bs4 import BeautifulSoup
def search_google(query: str, num_results: int = 5) -> str: """Search Google using SerpAPI.""" api_key = os.getenv("SERPAPI_KEY") url = "https://serpapi.com/search" params = { "q": query, "api_key": api_key, "num": num_results } try: response = requests.get(url, params=params) data = response.json() results = data.get("organic_results", []) output = "" for i, result in enumerate(results, 1): output += f"{i}. {result['title']}\n" output += f" {result['link']}\n" output += f" {result.get('snippet', 'No snippet')}\n\n" return output if results else "No results found." except Exception as e: return f"Search error: {e}"
def read_article(url: str) -> str: """Read and extract the main text from a web article.""" try: response = requests.get(url, timeout=15) soup = BeautifulSoup(response.text, 'html.parser') # Remove script and style elements for element in soup(["script", "style", "nav", "header", "footer"]): element.decompose() # Get main content article = soup.find("article") or soup.find("main") or soup.find("body") text = article.get_text(separator="\n", strip=True) # Clean up whitespace lines = [line.strip() for line in text.splitlines() if line.strip()] clean_text = "\n".join(lines) return clean_text[:8000] # Limit to 8000 chars except Exception as e: return f"Error reading article: {e}"
# Define research tools research_tools = [ Tool(name="search_google", func=search_google, description="Search Google for information"), Tool(name="read_article", func=read_article, description="Read and extract text from a web article URL"), Tool(name="calculator", func=calculator, description="Perform mathematical calculations") ]
# Research agent prompt research_prompt = ChatPromptTemplate.from_messages([ ("system", """You are an expert research assistant. Your job is to:
- Search for relevant information using web search
- Read the most promising articles in detail
- Synthesize the information into a thorough report
- Cite your sources
Format your reports with:
- Executive Summary
- Key Findings (with source citations)
- Analysis
- Recommendations
Always search for multiple perspectives and verify claims across sources."""), MessagesPlaceholder(variable_name="chat_history", optional=True), ("human", "{input}"), MessagesPlaceholder(variable_name="agent_scratchpad") ])
# Create the research agent llm = ChatOpenAI(model="gpt-4o", temperature=0.3) research_agent = create_tool_calling_agent(llm, research_tools, research_prompt) research_executor = AgentExecutor( agent=research_agent, tools=research_tools, verbose=True, max_iterations=15 )
# Run a research task result = research_executor.invoke({ "input": "Research the current state of AI agent frameworks in 2025. Compare LangChain, AutoGen, CrewAI, and LangGraph." })
print(result["output"])
`
Example 2: Data Analysis Agent
`python
"""
Data Analysis Agent
Loads data, performs analysis, creates visualizations, and generates insights.
"""
import pandas as pd import matplotlib.pyplot as plt import io import base64
@tool def load_csv(filepath: str) -> str: """Load a CSV file and return a summary of its contents.""" try: df = pd.read_csv(filepath) summary = f"File: {filepath}\n" summary += f"Shape: {df.shape[0]} rows × {df.shape[1]} columns\n" summary += f"Columns: {', '.join(df.columns.tolist())}\n" summary += f"Data types:\n{df.dtypes.to_string()}\n" summary += f"\nFirst 5 rows:\n{df.head().to_string()}\n" summary += f"\nBasic statistics:\n{df.describe().to_string()}\n" return summary except Exception as e: return f"Error loading CSV: {e}"
@tool def analyze_data(query: str) -> str: """Analyze the loaded data. The query describes what analysis to perform. Available data is in the pandas DataFrame 'df'.""" try: # This is a simplified version. in production, use a code execution sandbox # The agent would generate and execute pandas code return f"Analysis result for: {query}" except Exception as e: return f"Analysis error: {e}"
@tool def create_chart(chart_type: str, x_column: str, y_column: str, title: str) -> str: """Create a chart from the loaded data. chart_type can be 'bar', 'line', 'scatter', or 'histogram'.""" try: df = pd.read_csv("temp_data.csv") # Assume data was saved temporarily plt.figure(figsize=(10, 6)) if chart_type == "bar": df.plot.bar(x=x_column, y=y_column) elif chart_type == "line": df.plot.line(x=x_column, y=y_column) elif chart_type == "scatter": df.plot.scatter(x=x_column, y=y_column) elif chart_type == "histogram": df[x_column].plot.hist() plt.title(title) plt.tight_layout() # Save to base64 string buffer = io.BytesIO() plt.savefig(buffer, format='png') buffer.seek(0) image_base64 = base64.b64encode(buffer.read()).decode() plt.close() return f"Chart created successfully. Base64: {image_base64[:100]}..." except Exception as e: return f"Chart error: {e}"
data_tools = [load_csv, analyze_data, create_chart]
data_prompt = ChatPromptTemplate.from_messages([ ("system", """You are a data analysis expert. You can:
- Load CSV files and examine their structure
- Perform statistical analysis and data transformations
- Create visualizations (bar charts, line plots, scatter plots, histograms)
- Generate insights and recommendations based on the data
Always start by loading and understanding the data before analyzing."""), ("human", "{input}"), MessagesPlaceholder(variable_name="agent_scratchpad") ])
data_agent = create_tool_calling_agent(llm, data_tools, data_prompt) data_executor = AgentExecutor(agent=data_agent, tools=data_tools, verbose=True)
`
Example 3: Customer Support Agent
`python
"""
Customer Support Agent
Handles customer inquiries with access to order data and knowledge base.
"""
@tool def lookup_order(order_id: str) -> str: """Look up an order by ID and return order details.""" # In production, this would query your database orders_db = { "ORD-001": {"status": "shipped", "item": "Wireless Headphones", "price": 79.99, "tracking": "1Z999AA10123456784"}, "ORD-002": {"status": "processing", "item": "USB-C Hub", "price": 45.99, "tracking": None}, "ORD-003": {"status": "delivered", "item": "Mechanical Keyboard", "price": 129.99, "tracking": "1Z999AA10123456785"}, } order = orders_db.get(order_id) if order: return json.dumps(order, indent=2) return f"Order {order_id} not found."
@tool def search_knowledge_base(query: str) -> str: """Search the company knowledge base for relevant articles.""" # In production, use a vector database (Pinecone, Weaviate, etc.) kb = { "return policy": "Items can be returned within 30 days of purchase with original receipt.", "shipping": "Standard shipping takes 5-7 business days. Express shipping takes 2-3 days.", "warranty": "All electronics come with a 1-year manufacturer warranty.", "payment": "We accept Visa, Mastercard, PayPal, and Apple Pay.", } results = [] for key, value in kb.items(): if query.lower() in key.lower() or query.lower() in value.lower(): results.append(f"{key}: {value}") return "\n".join(results) if results else "No relevant articles found."
@tool def create_ticket(customer_email: str, issue: str, priority: str) -> str: """Create a support ticket for issues that can't be resolved automatically.""" ticket_id = f"TKT-{hash(customer_email + issue) % 100000:05d}" return f"Ticket {ticket_id} created. Priority: {priority}. An agent will follow up within 24 hours."
support_tools = [lookup_order, search_knowledge_base, create_ticket]
support_prompt = ChatPromptTemplate.from_messages([ ("system", """You are a friendly customer support agent for TechStore Inc.
Guidelines:
- Always be polite and professional
- Look up order information when customers ask about their orders
- Search the knowledge base for policy questions
- Create a support ticket for issues you can't resolve
- If you need to create a ticket, ask for the customer's email first
Never make up information. If you don't know something, search the knowledge base or create a ticket."""), MessagesPlaceholder(variable_name="chat_history", optional=True), ("human", "{input}"), MessagesPlaceholder(variable_name="agent_scratchpad") ])
support_agent = create_tool_calling_agent( ChatOpenAI(model="gpt-4o", temperature=0.2), support_tools, support_prompt ) support_executor = AgentExecutor( agent=support_agent, tools=support_tools, memory=ConversationBufferMemory(memory_key="chat_history", return_messages=True), verbose=True )
# Example interaction result = support_executor.invoke({ "input": "Hi, I ordered wireless headphones last week and they haven't arrived yet. My order number is ORD-001." }) print(result["output"])
`
Deploying with Render
Now that our agent works locally, let's deploy it to the cloud using Render so anyone can access it via an API.
Step 1: Create a FastAPI Server
Create
main.py:
`python
from fastapi import FastAPI, HTTPException
from pydantic import BaseModel
from typing import Optional
import uvicorn
from agent import agent_executor # Import your agent
app = FastAPI( title="AI Agent API", description="A powerful AI agent with tool-calling capabilities", version="1.0.0" )
class AgentRequest(BaseModel): message: str session_id: Optional[str] = "default"
class AgentResponse(BaseModel): response: str session_id: str
@app.post("/chat", response_model=AgentResponse) async def chat(request: AgentRequest): """Send a message to the AI agent and get a response.""" try: result = agent_executor.invoke({ "input": request.message }) return AgentResponse( response=result["output"], session_id=request.session_id ) except Exception as e: raise HTTPException(status_code=500, detail=str(e))
@app.get("/health") async def health_check(): """Health check endpoint.""" return {"status": "healthy", "service": "ai-agent-api"}
if __name__ == "__main__": uvicorn.run(app, host="0.0.0.0", port=8000)
`
Step 2: Create a Web Interface (Optional)*
Create
templates/index.html:
`html
AI Agent Chat
🤖 AI Agent Chat
`
Add the template serving to
main.py:
`python
from fastapi.staticfiles import StaticFiles
from fastapi.templating import Jinja2Templates
from fastapi.responses import HTMLResponse
from fastapi import Request
templates = Jinja2Templates(directory="templates")
@app.get("/", response_class=HTMLResponse) async def home(request: Request): return templates.TemplateResponse("index.html", {"request": request})
`
Step 3: Prepare for Render Deployment
Create a
render.yaml file for Render's Blueprint deployment:
`yaml
services:
- type: web
name: ai-agent-api
env: python
buildCommand: pip install -r requirements.txt
startCommand: uvicorn main:app --host 0.0.0.0 --port $PORT
envVars:
- key: OPENAI_API_KEY
sync: false # Set manually in Render dashboard
- key: MODEL_NAME
value: gpt-4o
- key: TEMPERATURE
value: "0.7"
healthCheckPath: /health
`
Update
requirements.txt with all production dependencies:
`text
openai==1.35.0
langchain==0.2.10
langchain-openai==0.1.17
langchain-community==0.2.9
langgraph==0.1.1
python-dotenv==1.0.0
fastapi==0.111.0
uvicorn==0.30.1
pydantic==2.7.4
requests==2.32.3
beautifulsoup4==4.12.3
jinja2==3.1.4
`
Create a
runtime.txt for Python version:
`
python-3.11.9
`
Step 4: Deploy to Render
- Push to GitHub: Create a repository and push your code.
`bash
git init
git add .
git commit -m "Initial commit: AI Agent API"
git remote add origin https://github.com/yourusername/ai-agent-api.git
git push -u origin main
`
- Create a Render account at [render.com](https://render.com)
- Create a new Web Service:
Click "New +" → "Web Service"
Connect your GitHub repository
Render will auto-detect the render.yaml blueprint
Set your OPENAI_API_KEY as a secret environment variable in the Render dashboard
- Deploy: Click "Create Web Service" and Render will:
Install dependencies
Build the application
Start the server
Provide a public URL (e.g., https://ai-agent-api.onrender.com)
Step 5: Test the Deployment
`bash
# Test the health endpoint
curl https://your-app.onrender.com/health
# Test the chat endpoint curl -X POST https://your-app.onrender.com/chat \ -H "Content-Type: application/json" \ -d '{"message": "What is 25% of 840?"}'
`
Step 6: Set Up a Custom Domain (Optional)
In the Render dashboard:
- Go to your service → "Settings" → "Custom Domain"
- Add your domain (e.g.,
agent.yourdomain.com) Update your DNS records to point to Render Render automatically provisions SSL certificates
Next Steps and Best Practices
Security Best Practices
- Never expose API keys: Use environment variables and Render's secret management
- Rate limiting: Add rate limiting to prevent abuse
`python
from slowapi import Limiter
from slowapi.util import get_remote_address
limiter = Limiter(key_func=get_remote_address)
app.state.limiter = limiter
@app.post("/chat")
@limiter.limit("10/minute")
async def chat(request: AgentRequest):
...
`
- Input validation: Always validate and sanitize user inputs
- Tool sandboxing: Run tool executions in sandboxed environments
- Logging: Log all interactions for debugging and auditing
Performance Optimization
- Caching: Cache frequent queries and tool results
`python
from functools import lru_cache
import hashlib
@lru_cache(maxsize=100)
def cached_search(query: str) -> str:
return search_google(query)
`
- Async processing: Use async/await for I/O-bound operations
- Model selection: Use cheaper models (GPT-4o-mini) for simple tasks
- Streaming: Stream responses for better perceived performance
Monitoring and Observability
`python
import logging
import time
logging.basicConfig(level=logging.INFO) logger = logging.getLogger(__name__)
@app.post("/chat") async def chat(request: AgentRequest): start_time = time.time() try: result = agent_executor.invoke({"input": request.message}) duration = time.time() - start_time logger.info(f"Request processed in {duration:.2f}s | Session: {request.session_id}") return AgentResponse(response=result["output"], session_id=request.session_id) except Exception as e: logger.error(f"Error processing request: {e}") raise HTTPException(status_code=500, detail=str(e))
`
Scaling Your Agent
As your agent grows in complexity, consider these architectural patterns:
- Multi-agent systems: Use LangGraph to coordinate multiple specialized agents
- Tool servers: Separate tools into microservices for better maintainability
- Vector databases: Use Pinecone or Weaviate for knowledge retrieval
- Message queues: Use Redis or RabbitMQ for async task processing
- Containerization: Use Docker for consistent deployments
Going Further with LangGraph
For production-grade agents, LangGraph provides more control over the agent loop:
`python
from langgraph.graph import StateGraph, END
from typing import TypedDict, Annotated
import operator
class AgentState(TypedDict): messages: Annotated[list, operator.add] next_step: str tool_result: str
def should_continue(state: AgentState): """Decide whether to continue or end.""" last_message = state["messages"][-1] if last_message.get("tool_calls"): return "use_tool" return "respond"
def call_model(state: AgentState): """Call the LLM with current messages.""" response = llm.invoke(state["messages"]) return {"messages": [response]}
def call_tool(state: AgentState): """Execute tool calls from the last message.""" last_message = state["messages"][-1] results = [] for tool_call in last_message.tool_calls: # Execute tool and return result result = execute_tool(tool_call) results.append(result) return {"messages": results}
# Build the graph workflow = StateGraph(AgentState) workflow.add_node("agent", call_model) workflow.add_node("tools", call_tool)
workflow.set_entry_point("agent") workflow.add_conditional_edges("agent", should_continue) workflow.add_edge("tools", "agent")
app = workflow.compile()
``
Summary
In this tutorial, we covered the complete process of building and deploying an AI agent:
- AI Agent Architecture: Understanding the reasoning loop, tools, memory, and planning
- OpenAI API Integration: Making API calls, streaming, function calling, and structured output
- LangChain Framework: Using prompts, chains, output parsers, and the agent framework
- Agent Logic: Building ReAct agents from scratch and with LangChain's built-in tools
- Memory: Adding conversation memory with buffer, windowed, and summary-based approaches
- Tools: Creating custom tools for file operations, API calls, web search, and databases
- Real-World Examples: Research assistant, data analysis, and customer support agents
- Deployment: Creating a FastAPI server and deploying to Render with a web interface
The key takeaway is that modern AI agents combine the reasoning power of LLMs with the ability to use tools, remember context, and take actions. creating systems that can autonomously handle complex, multi-step tasks.
This tutorial is based on the freeCodeCamp course "Build Your Own AI Agent. Full Course with OpenAI, Langchain, Render Deployment" by Code with Ania Kubów.