Learn how to build a custom AI agent from scratch. Complete guide covering agent architecture, tool integration, memory systems, and deployment. Includes code examples for LangChain, AutoGPT, and custom implementations.
AI agents are no longer science fiction—they're practical tools you can build yourself. Unlike simple chatbots that respond to prompts, AI agents can plan, use tools, maintain memory, and execute multi-step tasks autonomously. This guide shows you how to build your own from scratch.
Whether you want an agent to manage your emails, research topics, automate workflows, or assist with coding, this comprehensive tutorial covers everything from architecture to deployment.
Traditional Chatbot:
AI Agent:
Every AI agent has four essential components:
The ReAct pattern alternates between reasoning and action:
```
Thought: I need to find the current weather
Action: search_weather("San Francisco")
Observation: 68°F, sunny
Thought: Now I can answer the user
Final Answer: It's 68°F and sunny in San Francisco
```
Best for: General-purpose agents, research tasks, multi-step workflows
Agent creates a complete plan first, then executes:
```
Plan:
Execute: [runs each step sequentially]
```
Best for: Complex tasks with clear goals, report generation, data analysis
Agent evaluates its own outputs and iterates:
```
Action: Generate code
Critique: Code has bug in line 15
Action: Fix bug
Critique: Looks good now
Final: Return corrected code
```
Best for: Code generation, content creation, quality-critical tasks
| Pattern | Complexity | Token Usage | Reliability | Best Use Case |
|---------|-----------|-------------|-------------|---------------|
| ReAct | Low | Medium | High | General tasks, research |
| Plan-and-Execute | Medium | High | Medium | Multi-step workflows |
| Reflexion | High | Very High | Very High | Quality-critical work |
| Hybrid | High | High | High | Production systems |
Prerequisites:
Python Setup:
```bash
python -m venv agent-env
source agent-env/bin/activate # On Windows: agent-env\Scripts\activate
pip install langchain langchain-openai langchain-anthropic
pip install chromadb # For memory
pip install duckduckgo-search # For web search
pip install python-dotenv
```
Node.js Setup:
```bash
npm init -y
npm install langchain @langchain/openai @langchain/anthropic
npm install chromadb # For memory
npm install axios cheerio # For web tools
npm install dotenv
```
Environment Configuration:
```bash
OPENAI_API_KEY=sk-xxx
ANTHROPIC_API_KEY=sk-ant-xxx
OLLAMA_BASE_URL=http://localhost:11434
```
Simple ReAct Agent:
```python
from langchain.agents import AgentExecutor, create_react_agent
from langchain_openai import ChatOpenAI
from langchain.tools import Tool
from langchain import hub
import os
from dotenv import load_dotenv
load_dotenv()
llm = ChatOpenAI(
model="gpt-4-turbo-preview",
temperature=0.7,
api_key=os.getenv("OPENAI_API_KEY")
)
def calculator(ion: str) -> str:
"""Evaluates mathematical expressions"""
try:
return str(eval(expression))
except Exception as e:
return f"Error: {str(e)}"
def web_search(query: str) -> str:
"""Searches the web for information"""
from duckduckgo_search import DDGS
results = DDGS().text(query, max_results=3)
return "\n".join([f"{r['title']}: {r['body']}" for r in results])
tools = [
Tool(
name="Calculator",
func=calculator,
description="Useful for mathematical calculations. Input should be a valin expression."
),
Tool(
name="WebSearch",
func=web_search,
description="Searches the web for current information. Input should be a search query."
)
]
prompt = hub.pull("hwchase17/react")
agent = create_react_agent(llm, tools, prompt)
agent_executor = AgentExecutor(
agent=agent,
tools=tools,
verbose=True,
max_iterations=5,
handle_parsing_errors=True
)
result = agent_executor.invoke({
"input": "What's the current price of Bitcoin multiplied by 100?"
})
print(result["output"])
```
Conversation Memory:
```python
from langchain.memory import ConversationBufferMemory
from langchain.agents import AgentExecutor, create_react_agent
memory = ConversationBufferMemory(
memory_key="chat_history",
return_messages=True
)
agent_executor = AgentExecutor(
agent=agent,
tools=tools,
memory=memory,
verbose=True
)
agent_executor.invoke({"input": "My name is Alice"})
agent_executor.invoke({"input": "What's my name?"}) # Returns: Alice
```
Long-Term Memory with Vector Store:
```python
from langchain.vectorstores import Chroma
from langchain.embeddings import OpenAIEmbeddings
from langchain.tools import Tool
embeddings = OpenAIEmbeddings()
vectorstore = Chroma(
collection_name="agent_memory",
embedding_function=embeddings,
persist_directory="./agent_db"
)
def remember(text: str) -> str:
"""Stores information in long-term memory"""
vectorstore.add_texts([text])
return "Information stored successfully"
def recall(query: str) -> str:
"""Retrieves information from long-term memory"""
docs = vectorstore.similarity_search(query, k=3)
return "\n".join([doc.page_content for doc in docs])
memory_tools = [
Tool(name="Remember", func=remember, description="Store information for later"),
Tool(name="Recall", func=recall, description="Retrieve stored information")
]
tools.extend(memory_tools)
```
File System Tools:
```python
import os
from pathlib import Path
def read_file(filepath: str) -> str:
"""Reads content from a file"""
try:
with open(filepath, 'r') as f:
return f.read()
except Exception as e:
return f"Error reading file: {str(e)}"
def write_file(filepath: str, content: str) -> str:
"""Writes content to a file"""
try:
Path(filepath).parent.mkdir(parents=True, exist_ok=True)
with open(filepath, 'w') as f:
f.write(content)
return f"Successfully wrote to {filepath}"
except Exception as e:
return f"Error writing file: {str(e)}"
def list_files(directory: str = ".") -> str:
"""Lists files in a directory"""
try:
files = os.listdir(directory)
return "\n".join(files)
except Exception as e:
return f"Error listing files: {str(e)}"
```
API Integration Tools:
```python
import requests
def fetch_api(url: str, method: str = "GET", data: dict = None) -> str:
"""Makes HTTP requests to APIs"""
try:
if method == "GET":
response = requests.get(url)
elif method == "POST":
response = requests.post(url, json=data)
return response.text
except Exception as e:
return f"API Error: {str(e)}"
def send_email(to: str, subject: str, body: str) -> str:
"""Sends email via API (example with SendGrid)"""
# Implementation depends on your email service
return f"Email sent to {to}"
```
Code Execution Sandbox:
```python
import subprocess
import tempfile
def execute_python(code: str) -> str:
"""Executes Python code in a safe sandbox"""
try:
with tempfile.NamedTemporaryFile(mode='w', suffix='.py', delete=False) as f:
f.write(code)
temp_file = f.name
result = subprocess.run(
['python', temp_file],
capture_output=True,
text=True,
timeout=5
)
os.unlink(temp_file)
return result.stdout if result.returncode == 0 else result.stderr
except Exception as e:
return f"Execution error: {str(e)}"
```
```python
from langchain.experimental import AutoGPT
from langchain.chat_models import ChatOpenAI
from langchain.tools import DuckDuckGoSearchRun
llm = ChatOpenAI(model="gpt-4", temperature=0.7)
search = DuckDuckGoSearchRun()
agent = AutoGPT.from_llm_and_tools(
ai_name="ResearchAgent",
le="Research assistant",
tools=[search],
llm=llm,
memory=vectorstore.as_retriever()
)
agent.run(["Research the latest AI developments and create a summary report"])
```
```python
class SimpleAgent:
def __init__(self, llm, tools):
self.llm = llm
self.tools = {tool.name: tool for tool in tools}
self.memory = []
def run(self, task: str, max_iterations: int = 5):
self.memory.append(f"Task: {task}")
for i in range(max_iterations):
# Get next action from LLM
prompt = self._build_prompt()
response = self.llm.predict(prompt)
# Parse action
action, action_input = self._parse_action(response)
if action == "Final Answer":
return action_input
# Execute tool
if action in self.tools:
observation = self.tools[action].func(action_input)
self.memory.append(f"Action: {action}({action_input})")
self.memory.append(f"Observation: {observation}")
else:
self.memory.append(f"Error: Unknown action {action}")
return "Max iterations reached"
def _build_prompt(self):
history = "\n".join(self.memory)
return f"""You are an AI agent. Use tools to complete tasks.
Available tools:
{self._format_tools()}
History:
{history}
What's your next action? Format: Action: [tool_name]
Action Input: [input]
Or: Final Answer: [answer]
"""
def _format_tools(self):
return "\n".join([f"- {name}: {tool.description}"
for name, tool in self.tools.items()])
def _parse_action(self, response: str):
# Simple parsing logic
if "Final Answer:" in response:
return "Final Answer", response.split("Final Answer:")[1].strip()
lines = response.split("\n")
action = lines[0].replace("Action:", "").strip()
action_input = lines[1].replace("Action Input:", "").strip()
return action, action_input
```
Run as CLI Tool:
```python
import sys
from agent import agent_executor
if __name__ == "__main__":
task = " ".join(sys.argv[1:])
result = agent_executor.invoke({"input": task})
print(result["output"])
```
```bash
python agent_cli.py "Research AI news and summarize"
```
Run as Background Service:
```python
import schedule
import time
def daily_report():
result = agent_executor.invoke({
"input": "Generate daily AI news summary and email to me"
})
print(result["output"])
schedule.every().day.at("09:00").do(daily_report)
while True:
schedule.run_pending()
time.sleep(60)
```
FastAPI Server:
```python
from fastapi import FastAPI, BackgroundTasks
from pydantic import BaseModel
app = FastAPI()
class AgentRequest(BaseModel):
task: str
user_id: str
@app.post("/agent/run")
async def run_agent(request: AgentRequest, background_tasks: BackgroundTasks):
# Run agent in background
background_tasks.add_task(execute_agent, request.task, request.user_id)
return {"status": "started", "task": request.task}
def execute_agent(task: str, user_id: str):
result = agent_executor.invoke({"input": task})
# Store result or send notification
print(f"Task completed for {user_id}: {result['output']}")
```
Docker Container:
```dockerfile
FROM python:3.10-slim
WORKDIR /app
COPY requirements.txt .
RUN pip install --no-cache-dir -r requirements.txt
COPY . .
CMD ["python", "agent_service.py"]
```
Deploy to Railway/Render:
```bash
npm install -g @railway/cli
railway login
railway init
railway up
```
1. Use Cheaper Models for Simple Tasks:
```python
from langchain.chat_models import ChatOpenAI
cheap_llm = ChatOpenAI(model="gpt-3.5-turbo", temperature=0.7)
expensive_llm = ChatOpenAI(model="gpt-4-turbo-preview", temperature=0.7)
def smart_agent(task: str):
# Classify task complexity
if is_simple_task(task):
return cheap_llm.predict(task)
else:
return expensive_llm.predict(task)
```
2. Implement Caching:
```python
from functools import lru_cache
import hashlib
@lru_cache(maxsize=1000)
def cached_llm_call(prompt: str):
return llm.predict(prompt)
import redis
r = redis.Redis(host='localhost', port=6379, db=0)
def cached_agent_call(task: str):
cache_key = hashlib.md5(task.encode()).hexdigest()
cached = r.get(cache_key)
if cached:
return cached.decode()
result = agent_executor.invoke({"input": task})
r.setex(cache_key, 3600, result["output"]) # Cache for 1 hour
return result["output"]
```
3. Use Local Models for Privacy/Cost:
```python
from langchain.llms import Ollama
local_llm = Ollama(model="llama2", base_url="http://localhost:11434")
def hybrid_agent(task: str):
# Draft with local model (free)
draft = local_llm.predict(f"Draft response: {task}")
# Refine with cloud model (paid, but less tokens)
refined = llm.predict(f"Improve this response: {draft}")
return refined
```
| Usage Level | Model | Tokens/Month | Cost |
|-------------|-------|--------------|------|
| Light (10 tasks/day) | GPT-3.5 | 300K | $0.60 |
| Medium (50 tasks/day) | GPT-3.5 | 1.5M | $3.00 |
| Heavy (50 tasks/day) | GPT-4 | 1.5M | $45.00 |
| Hybrid (50 tasks/day) | GPT-3.5 + GPT-4 | 1M + 500K | $17.00 |
| Local (unlimited) | Llama 2 (Ollama) | Unlimited | $0 (hardware only) |
Never hardcode keys:
```python
llm = ChatOpenAI(api_key="sk-xxx")
import os
from dotenv import load_dotenv
load_dotenv()
llm = ChatOpenAI(api_key=os.getenv("OPENAI_API_KEY"))
```
Use key rotation:
```python
import os
from datetime import datetime, timedelta
class RotatingAPIKey:
def __init__(self):
self.keys = [
os.getenv("OPENAI_KEY_1"),
os.getenv("OPENAI_KEY_2")
]
self.current_index = 0
self.last_rotation = datetime.now()
def get_key(self):
# Rotate every 7 days
if datetime.now() - self.last_rotation > timedelta(days=7):
self.current_index = (self.current_index + 1) % len(self.keys)
self.last_rotation = datetime.now()
return self.keys[self.current_index]
```
Local-first architecture:
```python
from langchain.vectorstores import Chroma
from langchain.embeddings import HuggingFaceEmbeddings
embeddings = HuggingFaceEmbeddings(model_name="all-MiniLM-L6-v2")
vectorstore = Chroma(
embedding_function=embeddings,
persist_directory="./local_memory" # Local storage
)
```
Sanitize inputs:
```python
import re
def sanitize_input(text: str) -> str:
# Remove potential injection attempts
text = re.sub(r'
text = re.sub(r'javascript:', '', text, flags=re.IGNORECASE)
return text.strip()
def safe_agent_call(user_input: str):
clean_input = sanitize_input(user_input)
return agent_executor.invoke({"input": clean_input})
```
Problem: Agent keeps repeating same actions
Solution: Add iteration limits and loop detection
```python
agent_executor = AgentExecutor(
agent=agent,
tools=tools,
max_iterations=5, # Limit iterations
early_stopping_method="generate" # Stop on repeated actions
)
```
Problem: Agent can't parse tool outputs
Solution: Improve tool descriptions and output formatting
```python
Tool(
name="Calculator",
func=calculator,
description="Calculates math expressions. Input: '2+2'. Output: '4'. Always returns a number."
)
```
Problem: Agent uses too many tokens
Solution: Implement token tracking and limits
```python
from langchain.callbacks import get_openai_callback
with get_openai_callback() as cb:
result = agent_executor.invoke({"input": task})
print(f"Tokens used: {cb.total_tokens}")
print(f"Cost: ${cb.total_cost:.4f}")
```
You now have a fully functional AI agent. Here's how to expand:
The OpenClaw Team specializes in AI infrastructure and agent development. We help individuals and businesses build custom AI solutions. Our open-source projects have helped thousands deploy their own AI systems.
Need help building your agent? Get a free AI audit to discuss your use case.
---
Ready to build your agent? Start with the code examples above, or contact us for personalized guidance.
Complete guide to ClawHub, the collaborative platform for AI agents. Learn how to discover, share, and deploy agents with detailed tutorials, best practices, and real-world examples.
Curated collection of the most popular Claude Code skills on GitHub. Discover productivity-boosting skills, implementation examples, and best practices for extending your AI coding assistant.
Step-by-step guide to installing and deploying OpenClaw AI agent framework. Learn how to set up your personal AI assistant with detailed configuration examples and troubleshooting tips.
Get your free AI audit and discover optimization opportunities.
START FREE AUDIT