Building an AI Agent in Python: From Zero to Production

AI agents that use tools, maintain memory, and handle complex tasks are transforming automation. This guide builds a complete agent system from scratch with production-grade reliability.

What You'll Build

A fully functional AI agent that can use external tools, maintain conversation memory, handle errors gracefully, and execute multi-step tasks. You'll learn the patterns behind production AI agent systems.

Why Build AI Agents?

AI agents go beyond simple chatbots:

• Tool use — agents call APIs, run code, and interact with systems
• Memory — maintain context across conversations
• Planning — break complex tasks into steps
• Autonomy — make decisions without human intervention

Full Tutorial

Step 1: Project Structure

ai-agent/
├── agent/
│   ├── __init__.py
│   ├── core.py
│   ├── llm.py
│   ├── tools.py
│   ├── memory.py
│   └── planner.py
├── config.yaml
├── requirements.txt
└── main.py

pip install openai pydantic pyyaml tiktoken

Step 2: LLM Interface

# agent/llm.py
from openai import AsyncOpenAI
from pydantic import BaseModel
from typing import Optional
import tiktoken

class LLMConfig(BaseModel):
    model: str = "gpt-4"
    temperature: float = 0.7
    max_tokens: int = 4096
    api_key: str

class LLMInterface:
    def __init__(self, config: LLMConfig):
        self.client = AsyncOpenAI(api_key=config.api_key)
        self.config = config
        self.encoding = tiktoken.encoding_for_model(config.model)

    def count_tokens(self, text: str) -> int:
        return len(self.encoding.encode(text))

    async def chat(self, messages: list[dict], tools: list[dict] = None) -> dict:
        kwargs = {
            "model": self.config.model,
            "messages": messages,
            "temperature": self.config.temperature,
            "max_tokens": self.config.max_tokens,
        }

        if tools:
            kwargs["tools"] = tools

        response = await self.client.chat.completions.create(**kwargs)

        return {
            "content": response.choices[0].message.content,
            "tool_calls": response.choices[0].message.tool_calls,
            "usage": {
                "prompt_tokens": response.usage.prompt_tokens,
                "completion_tokens": response.usage.completion_tokens,
            }
        }

Step 3: Tool System

# agent/tools.py
from typing import Callable, Any
from pydantic import BaseModel
import json
import inspect

class ToolDefinition(BaseModel):
    name: str
    description: str
    parameters: dict
    function: Callable

class ToolRegistry:
    def __init__(self):
        self.tools: dict[str, ToolDefinition] = {}

    def register(self, name: str, description: str, parameters: dict):
        def decorator(func: Callable):
            self.tools[name] = ToolDefinition(
                name=name,
                description=description,
                parameters=parameters,
                function=func
            )
            return func
        return decorator

    def get_openai_tools(self) -> list[dict]:
        tools = []
        for tool in self.tools.values():
            tools.append({
                "type": "function",
                "function": {
                    "name": tool.name,
                    "description": tool.description,
                    "parameters": tool.parameters
                }
            })
        return tools

    async def execute(self, name: str, arguments: str) -> str:
        if name not in self.tools:
            return json.dumps({"error": f"Tool {name} not found"})

        tool = self.tools[name]
        try:
            args = json.loads(arguments)
            result = await tool.function(**args)
            return json.dumps({"result": result})
        except Exception as e:
            return json.dumps({"error": str(e)})

registry = ToolRegistry()

@registry.register(
    name="calculate",
    description="Perform mathematical calculations",
    parameters={
        "type": "object",
        "properties": {
            "expression": {"type": "string", "description": "Math expression to evaluate"}
        },
        "required": ["expression"]
    }
)
async def calculate(expression: str) -> float:
    allowed_chars = set("0123456789+-*/.() ")
    if not all(c in allowed_chars for c in expression):
        raise ValueError("Invalid characters in expression")
    return eval(expression)

@registry.register(
    name="web_search",
    description="Search the web for information",
    parameters={
        "type": "object",
        "properties": {
            "query": {"type": "string", "description": "Search query"}
        },
        "required": ["query"]
    }
)
async def web_search(query: str) -> str:
    # Replace with actual search API
    return f"Search results for: {query}"

@registry.register(
    name="run_python",
    description="Execute Python code safely",
    parameters={
        "type": "object",
        "properties": {
            "code": {"type": "string", "description": "Python code to execute"}
        },
        "required": ["code"]
    }
)
async def run_python(code: str) -> str:
    # In production, use a sandboxed environment
    local_vars = {}
    exec(code, {"__builtins__": {}}, local_vars)
    return str(local_vars)

Step 4: Memory System

# agent/memory.py
from datetime import datetime
from dataclasses import dataclass, field
from typing import Optional
import json

@dataclass
class MemoryEntry:
    content: str
    role: str
    timestamp: datetime = field(default_factory=datetime.now)
    metadata: dict = field(default_factory=dict)

class ConversationMemory:
    def __init__(self, max_entries: int = 100, summary_threshold: int = 50):
        self.entries: list[MemoryEntry] = []
        self.max_entries = max_entries
        self.summary_threshold = summary_threshold
        self.summary: Optional[str] = None

    def add(self, role: str, content: str, metadata: dict = None):
        entry = MemoryEntry(
            content=content,
            role=role,
            metadata=metadata or {}
        )
        self.entries.append(entry)

        if len(self.entries) > self.max_entries:
            self._compress()

    def get_messages(self, limit: int = 20) -> list[dict]:
        messages = []

        if self.summary:
            messages.append({
                "role": "system",
                "content": f"Previous conversation summary: {self.summary}"
            })

        recent = self.entries[-limit:]
        for entry in recent:
            messages.append({
                "role": entry.role,
                "content": entry.content
            })

        return messages

    def _compress(self):
        # Keep last 20 entries and summarize the rest
        to_summarize = self.entries[:-20]
        self.entries = self.entries[-20:]

        texts = [e.content for e in to_summarize if e.role == "user"]
        self.summary = f"User discussed: {'; '.join(texts[-5:])}"

    def save(self, filepath: str):
        data = {
            "entries": [
                {
                    "content": e.content,
                    "role": e.role,
                    "timestamp": e.timestamp.isoformat(),
                    "metadata": e.metadata
                }
                for e in self.entries
            ],
            "summary": self.summary
        }
        with open(filepath, "w") as f:
            json.dump(data, f, indent=2)

    def load(self, filepath: str):
        with open(filepath) as f:
            data = json.load(f)

        self.summary = data.get("summary")
        self.entries = [
            MemoryEntry(
                content=e["content"],
                role=e["role"],
                timestamp=datetime.fromisoformat(e["timestamp"]),
                metadata=e.get("metadata", {})
            )
            for e in data.get("entries", [])
        ]

Step 5: Agent Core

# agent/core.py
from agent.llm import LLMInterface, LLMConfig
from agent.tools import registry
from agent.memory import ConversationMemory
import json

class AIAgent:
    def __init__(self, config: LLMConfig):
        self.llm = LLMInterface(config)
        self.memory = ConversationMemory()
        self.tools = registry
        self.system_prompt = """You are a helpful AI assistant with access to tools.
Always think step by step. Use tools when needed to complete tasks.
Be concise and accurate in your responses."""

    async def process(self, user_input: str) -> str:
        self.memory.add("user", user_input)

        messages = [{"role": "system", "content": self.system_prompt}]
        messages.extend(self.memory.get_messages())

        tools = self.tools.get_openai_tools()
        max_iterations = 10

        for _ in range(max_iterations):
            response = await self.llm.chat(messages, tools)

            if response["tool_calls"]:
                messages.append({
                    "role": "assistant",
                    "tool_calls": response["tool_calls"]
                })

                for tool_call in response["tool_calls"]:
                    result = await self.tools.execute(
                        tool_call.function.name,
                        tool_call.function.arguments
                    )
                    messages.append({
                        "role": "tool",
                        "tool_call_id": tool_call.id,
                        "content": result
                    })
            else:
                final_response = response["content"]
                self.memory.add("assistant", final_response)
                return final_response

        return "I apologize, but I couldn't complete this task. Please try again."

    def save_state(self, filepath: str):
        self.memory.save(filepath)

    def load_state(self, filepath: str):
        self.memory.load(filepath)

Step 6: Main Application

# main.py
import asyncio
from agent.core import AIAgent
from agent.llm import LLMConfig
import os

async def main():
    config = LLMConfig(
        model="gpt-4",
        api_key=os.environ["OPENAI_API_KEY"],
        temperature=0.7
    )

    agent = AIAgent(config)

    print("AI Agent started! Type 'quit' to exit.")
    print("Available commands: /tools, /memory, /save\n")

    while True:
        user_input = input("You: ").strip()

        if user_input.lower() == "quit":
            break

        if user_input == "/tools":
            tools = agent.tools.get_openai_tools()
            for tool in tools:
                print(f"- {tool['function']['name']}: {tool['function']['description']}")
            continue

        if user_input == "/memory":
            messages = agent.memory.get_messages(limit=5)
            for msg in messages:
                print(f"[{msg['role']}] {msg['content'][:100]}...")
            continue

        if user_input == "/save":
            agent.save_state("agent_state.json")
            print("State saved!")
            continue

        response = await agent.process(user_input)
        print(f"\nAgent: {response}\n")

if __name__ == "__main__":
    asyncio.run(main())

Production Considerations

• Implement proper error handling and retries
• Add logging for debugging and monitoring
• Use async throughout for scalability
• Implement rate limiting for API calls
• Add user authentication for multi-tenant systems

Get the Code

Ready to use these tools? Browse our collection of tested, production-ready Python scripts:

🔗 Browse Products: Anna's Digital Products

All products include:

• ✅ Tested and verified code
• ✅ Instant delivery via crypto or card
• ✅ Free updates forever
• ✅ Telegram bot support (@AnnaLilithBot)