Ayush Kumar

Posted on Jul 14 • Originally published at logiclooptech.dev

How to Build AI Agents from Scratch with FastAPI

#aiagents #python #llm #tutorial

How do I set up the Python environment for building AI agents from scratch?

You can start building AI agents from scratch by getting a clean Python environment and installing the right libraries. I usually begin with pyenv or conda to keep the interpreter isolated, then spin up a virtual environment with venv. Here’s my typical setup on an Ubuntu box:

# Install a specific Python version (optional)
pyenv install 3.11.5
pyenv global 3.11.5

# Create and activate a virtual environment
python -m venv .venv
source .venv/bin/activate

# Upgrade pip and install core deps
pip install --upgrade pip
pip install fastapi uvicorn python-dotenv

For the AI side I add the OpenAI SDK, LangChain, LangGraph and a tiny SQLite wrapper for persistence:

pip install openai langchain langgraph[all] sqlalchemy

If you plan to run the agent on a GPU, add torch and transformers. Pin versions in a requirements.txt so the CI pipeline never surprises you:

fastapi==0.110.0
uvicorn[standard]==0.27.0
openai==1.12.0
langchain==0.2.0
langgraph==0.0.12
sqlalchemy==2.0.29
python-dotenv==1.0.1

What breaks most often?

Forgetting to activate the venv inside Docker builds.
Mismatched OpenAI SDK versions that change the chat completion signature.

I caught both issues early by adding a simple sanity test to the CI pipeline:

# tests/test_imports.py
def test_imports():
    import fastapi, openai, langchain, langgraph, sqlalchemy

If the test fails, the build stops before any heavy work.

Which agent framework should I pick: LangChain, LangGraph, or the OpenAI SDK?

The short answer is: pick the tool that matches the complexity of your agent. If you only need a thin wrapper around ChatCompletion, the OpenAI SDK is enough. When you need composable chains, tool selection, and a built-in memory abstraction, LangChain is a solid middle ground. If you want explicit graph-based control over tool-use loops, LangGraph gives you the most flexibility.

OpenAI SDK – minimal, low-overhead

import openai

def simple_chat(prompt: str) -> str:
    resp = openai.ChatCompletion.create(
        model="gpt-4o-mini",
        messages=[{"role": "user", "content": prompt}],
        temperature=0.2,
    )
    return resp.choices[0].message.content

Pros: tiny dependency footprint, easy to audit. Cons: you have to wire up memory, tool calling, and retries yourself.

LangChain – batteries-included for most use-cases

from langchain import OpenAI, LLMChain, PromptTemplate
from langchain.memory import ConversationBufferMemory

llm = OpenAI(model="gpt-4o-mini", temperature=0.2)
prompt = PromptTemplate.from_template("{history}\nUser: {input}\nAI:")
memory = ConversationBufferMemory(k=5)
chain = LLMChain(llm=llm, prompt=prompt, memory=memory)

def chat_with_memory(user_input: str) -> str:
    return chain.run({"input": user_input})

Pros: built-in memory, tool wrappers, easy to swap LLM providers. Cons: a larger dependency tree; sometimes the abstraction hides latency spikes.

LangGraph – explicit graph control

from langgraph.graph import GraphBuilder
from langchain import OpenAI
from langgraph.prebuilt import tool_node

llm = OpenAI(model="gpt-4o-mini")
builder = GraphBuilder()

@builder.node
def start(state):
    return {"msg": "What do you need help with?"}

@builder.node
def tool_use(state):
    # Example tool call: a simple calculator
    if "calc:" in state["msg"]:
        expr = state["msg"].split("calc:")[1]
        result = eval(expr)  # never do this in prod!
        return {"msg": f"The answer is {result}"}
    return {"msg": state["msg"]}

builder.add_edge("start", "tool_use")
graph = builder.compile()

Pros: you can see the exact flow, add loops, and debug each node. Cons: more boilerplate, steeper learning curve.

When NOT to use a heavy framework?

If the agent is a single-turn query or a low-traffic webhook, the OpenAI SDK keeps latency low and the Docker image small.

How do I design the agent architecture with tools, memory and tool-use loops?

The architecture boils down to three moving parts: the LLM core, a set of tools the model can invoke, and a memory layer that holds state between turns.

1. LLM core

Wrap the model behind a thin service so you can swap providers later. I expose a generate function that receives a list of messages and returns the assistant reply.

def generate(messages: list[dict]) -> str:
    resp = openai.ChatCompletion.create(
        model="gpt-4o-mini",
        messages=messages,
        temperature=0.0,
    )
    return resp.choices[0].message.content

2. Tool interface

Define a simple protocol that each tool must implement:

from typing import Protocol, Any

class Tool(Protocol):
    name: str
    description: str

    def run(self, input: str) -> Any:
        ...

Then register tools in a dict:

tools: dict[str, Tool] = {}

def register(tool: Tool):
    tools[tool.name] = tool

A concrete example – a weather lookup:

import httpx

class WeatherTool:
    name = "weather"
    description = "Fetches current weather for a city."

    async def run(self, city: str) -> str:
        url = f"https://api.open-meteo.com/v1/forecast?city={city}&current_weather=true"
        async with httpx.AsyncClient() as client:
            data = await client.get(url)
        return data.json()["current_weather"]["temperature"]  # simplified

3. Memory layer

I prefer a hybrid approach: short-term memory lives in Redis (fast, shared across workers) while long-term facts are stored in SQLite via SQLAlchemy. The short-term store holds the last k messages; the long-term store keeps user preferences.

# redis_mem.py
import redis
import json
r = redis.from_url("redis://localhost:6379/0")

def push_message(user_id: str, role: str, content: str):
    key = f"mem:{user_id}"
    r.rpush(key, json.dumps({"role": role, "content": content}))
    r.ltrim(key, -10, -1)  # keep last 10 messages

def get_history(user_id: str) -> list[dict]:
    key = f"mem:{user_id}"
    raw = r.lrange(key, 0, -1)
    return [json.loads(m) for m in raw]

4. The loop

Putting it together:

async def agent_loop(user_id: str, user_input: str) -> str:
    # 1. Load recent history
    history = get_history(user_id)

    # 2. Append current user message
    history.append({"role": "user", "content": user_input})

    # 3. Let LLM decide whether to call a tool
    response = generate(history)

    # Simple heuristic: if response contains a tool tag like "<tool:weather>"
    if response.startswith("<tool:"):
        tool_name, arg = response[6:].split(">", 1)
        tool = tools.get(tool_name)
        if tool:
            tool_result = await tool.run(arg.strip())
            # 4. Feed tool result back to LLM
            history.append({"role": "assistant", "content": f"Tool result: {tool_result}"})
            final_reply = generate(history)
        else:
            final_reply = "I don't know that tool."
    else:
        final_reply = response

    # 5. Persist final reply
    push_message(user_id, "assistant", final_reply)
    return final_reply

Trade-offs

Using Redis for short-term memory adds a network hop; if latency is critical, keep it in-process for single-worker setups.
Storing every turn in SQLite inflates the DB quickly; prune or archive old sessions.

What’s the best way to implement persistent memory and state management?

Persistent memory is the glue that makes an agent feel like a consistent assistant. I split state into three layers:

Layer	Purpose	Tech
Ephemeral (per request)	Prompt construction	Python objects
Short-term (last few turns)	Conversation context	Redis (TTL 24 h)
Long-term (user profile, preferences)	Knowledge base	SQLite + SQLAlchemy

SQLite schema example

# models.py
from sqlalchemy import Column, Integer, String, JSON, create_engine
from sqlalchemy.orm import declarative_base, sessionmaker

Base = declarative_base()

class UserProfile(Base):
    __tablename__ = "user_profiles"
    id = Column(Integer, primary_key=True)
    user_id = Column(String, unique=True, index=True)
    prefs = Column(JSON)  # e.g., {"language": "en", "units": "metric"}

engine = create_engine("sqlite:///agents.db", future=True)
SessionLocal = sessionmaker(bind=engine, expire_on_commit=False)

def init_db():
    Base.metadata.create_all(engine)

When the agent needs a piece of long-term info:

def get_user_prefs(user_id: str) -> dict:
    with SessionLocal() as db:
        profile = db.query(UserProfile).filter_by(user_id=user_id).first()
        return profile.prefs if profile else {}

Failure modes

Redis eviction can wipe short-term memory unexpectedly; set a reasonable maxmemory-policy.
SQLite writes lock under heavy concurrency; for >10 RPS consider Postgres or a file-based WAL setting.

When to skip persistence?

If the agent runs in a batch job (e.g., generate a report) and never talks to the same user twice, you can drop the long-term layer entirely.

How can I test, debug, and deploy the agent as a FastAPI service?

Testing an AI agent feels different from testing a CRUD endpoint because the LLM response is nondeterministic. I use three levels of testing:

Unit tests for tool logic and memory helpers (pure Python).
Integration tests that mock the OpenAI SDK with responses or pytest-mock.
End-to-end tests that spin up the FastAPI app with TestClient and run a few real prompts against a cheap model (e.g., gpt-4o-mini).

Unit test example

def test_redis_memory_roundtrip():
    user = "test_user"
    push_message(user, "user", "Hello")
    hist = get_history(user)
    assert len(hist) == 1
    assert hist[0]["content"] == "Hello"

Mocking the LLM

from unittest.mock import patch

@patch("openai.ChatCompletion.create")
def test_agent_calls_tool(mock_create):
    mock_create.return_value = SimpleNamespace(
        choices=[SimpleNamespace(message=SimpleNamespace(content="<tool:weather>London"))]
    )
    # Register a fake weather tool
    class FakeWeather:
        name = "weather"
        async def run(self, city): return "15°C"
    register(FakeWeather())

    reply = asyncio.run(agent_loop("u1", "What's the weather?"))
    assert "15°C" in reply

FastAPI endpoint

# main.py
from fastapi import FastAPI, HTTPException
from pydantic import BaseModel

app = FastAPI()

class ChatRequest(BaseModel):
    user_id: str
    message: str

@app.post("/chat")
async def chat(req: ChatRequest):
    try:
        reply = await agent_loop(req.user_id, req.message)
        return {"reply": reply}
    except Exception as e:
        raise HTTPException(status_code=500, detail=str(e))

Run locally with:

uvicorn main:app --reload

Deploying with Docker

# Dockerfile
FROM python:3.11-slim
WORKDIR /app
COPY requirements.txt .
RUN pip install --no-cache-dir -r requirements.txt
COPY . .
CMD ["uvicorn", "main:app", "--host", "0.0.0.0", "--port", "8000"]

I push the image to Docker Hub and deploy on Railway. The deployment pipeline is the same one I described in my post about FastAPI deployment on Railway. If you need more control over scaling, Render gives you a free tier with automatic TLS – see my notes on FastAPI deployment on Render.

Debugging tips

Log the full message list before each LLM call; it makes reproducing errors trivial.
Capture the raw OpenAI response headers; rate-limit info lives there.
If the container restarts repeatedly, check the memory limit – the model payload can exceed 2 GiB in edge cases.

Cost considerations

A single gpt-4o-mini call costs ~$0.00015. With 100 RPS you hit $13/day, not counting Redis and DB.
Turn on OpenAI’s logprobs only when you need fine-grained debugging; it adds latency.

FAQ

Can I use LangChain without Redis?

Yes. LangChain ships with an in-memory buffer that works fine for single-process development. Switch to Redis when you need horizontal scaling.

Do I need a GPU for the agent?

If you stay on OpenAI’s hosted models, no. The GPU cost is only relevant when you run a local LLM like Llama-3, which adds complexity and higher operational overhead.

What’s the easiest way to add new tools?

Implement the Tool protocol, register the instance with register(), and add a simple pattern in the loop to detect <tool:name> tags. No changes to the core LLM code are required.

How do I handle user authentication in the FastAPI endpoint?

Use OAuth2 with JWT tokens. FastAPI’s Depends system makes it straightforward to extract user_id from the token and pass it to agent_loop.

Key Takeaways

Set up an isolated Python environment; pin library versions to avoid sudden SDK breaks.
Choose the framework that matches your agent’s complexity: OpenAI SDK for minimal, LangChain for most, LangGraph for explicit graph control.
Separate short-term (Redis) and long-term (SQLite) memory to keep latency low and data durable.
Write unit and integration tests that mock the LLM; end-to-end tests catch prompt-level regressions.
Deploy with FastAPI in Docker; scale on Railway or Render, and monitor cost per token.

With these pieces in place you can reliably build AI agents from scratch, iterate quickly, and keep the production stack manageable. Happy coding.

DEV Community