Building an ai agent python langchain with FastAPI

What is a LangChain agent and how does its architecture work?

A LangChain agent is a wrapper around a language model that lets you define a loop of thought → action → observation.
In practice the agent receives a user prompt, asks the model to decide what to do next, executes the chosen tool (like a web search or a database query), feeds the result back to the model, and repeats until it produces a final answer.

The core pieces are:

In production I’ve seen agents choke when the LLM hallucinates a tool that doesn’t exist, or when the tool’s latency spikes. Guarding the executor with timeouts and a whitelist of allowed actions saves a lot of headaches.

How do I set up a Python environment and install LangChain?

First thing: isolate the stack. I use uvicorn for ASGI, python-3.11, and poetry for dependency management. The minimal pyproject.toml looks like this:

[tool.poetry]
name = "langchain-fastapi-agent"
version = "0.1.0"
description = "A FastAPI service exposing a LangChain AI agent"
authors = ["Your Name <you@example.com>"]
license = "MIT"
python = "^3.11"

[tool.poetry.dependencies]
fastapi = "^0.110.0"
uvicorn = {extras = ["standard"], version = "^0.27.0"}
langchain = "^0.1.0"
openai = "^1.3.0"
pydantic = "^2.5.0"
python-dotenv = "^1.0.0"

[build-system]
requires = ["poetry-core"]
build-backend = "poetry.core.masonry.api"

Run:

poetry install
poetry shell

Set your OpenAI key in a .env file:

OPENAI_API_KEY=sk-...

Why poetry? It pins exact versions, which is a lifesaver when a minor LangChain bump changes the tool-calling API. I’ve been bitten by a silent break after a pip install -U langchain that removed BaseTool in favor of Tool.

How can I build a simple AI agent with LangChain?

Let’s start with a “calculator” agent that can add, subtract, multiply, or divide numbers. The code below lives in agent.py.

import os
from langchain.llms import OpenAI
from langchain.agents import initialize_agent, Tool
from langchain.prompts import PromptTemplate
from langchain.memory import ConversationBufferMemory

# Load LLM
llm = OpenAI(model="gpt-4o-mini", temperature=0)

# Define a single tool
def calc(expression: str) -> str:
    """Evaluate a basic arithmetic expression and return the result."""
    try:
        result = eval(expression, {"__builtins__": {}})
        return str(result)
    except Exception as exc:
        return f"Error: {exc}"

calc_tool = Tool(
    name="Calculator",
    func=calc,
    description="Useful for evaluating simple arithmetic expressions, e.g. '2 + 2' or '12/4'."
)

# Memory keeps the conversation context
memory = ConversationBufferMemory(memory_key="chat_history")

# Prompt template – keep it short; long prompts increase token cost
prompt = PromptTemplate.from_template(
    """You are an assistant that can solve math problems.
    Use the provided tools when needed. Respond only with the final answer."""
)

# Build the agent
agent = initialize_agent(
    tools=[calc_tool],
    llm=llm,
    agent_type="zero-shot-react-description",
    memory=memory,
    prompt=prompt,
    verbose=True,
)

A quick test in the REPL:

>>> agent.run("What is 15 * 7?")
The answer is 105.

The zero-shot-react-description agent type is the most predictable for production because it doesn’t rely on few-shot examples that can drift over time.

Trade-offs

• Speed vs. cost – Every loop hit calls the LLM, so a 5-step reasoning chain can be pricey. If you only need deterministic math, a pure Python function is cheaper.
• Tool security – eval is dangerous. In the example I sandboxed it by removing __builtins__. In real services you should use a safe expression parser like asteval or numexpr.

How do I integrate the LangChain agent into a FastAPI application?

FastAPI gives us an async endpoint that forwards the user query to the agent. Create main.py:

import os
from fastapi import FastAPI, HTTPException
from pydantic import BaseModel
from dotenv import load_dotenv
from agent import agent

load_dotenv()  # pulls OPENAI_API_KEY into the environment

app = FastAPI(title="LangChain Agent Service")

class QueryRequest(BaseModel):
    prompt: str

class QueryResponse(BaseModel):
    answer: str

@app.post("/chat", response_model=QueryResponse)
async def chat(request: QueryRequest):
    try:
        # LangChain agents are sync; run it in a thread pool to avoid blocking the event loop
        from asyncio import to_thread
        answer = await to_thread(agent.run, request.prompt)
        return QueryResponse(answer=answer.strip())
    except Exception as exc:
        raise HTTPException(status_code=500, detail=str(exc))

Why to_thread?
I tried calling agent.run directly from an async route and the server stalled under load. Offloading to a thread pool keeps the ASGI loop responsive, at the cost of a small thread-creation overhead. For high QPS you can pre-warm a pool with concurrent.futures.ThreadPoolExecutor(max_workers=8) and reuse it.

Run locally:

uvicorn main:app --reload

POST to http://localhost:8000/chat with JSON { "prompt": "What is 23 * 42?" } and you’ll get the computed answer.

Failure modes

• Timeouts – If the LLM takes longer than your client’s patience, the request bubbles up as a 500. Wrap the call in asyncio.wait_for with a reasonable deadline (e.g., 10 s).
• Model rate limits – OpenAI will return a 429. Cache recent results or implement exponential back-off.

How can I add memory and tool-use capabilities to the agent?

The simple calculator only needed one tool. Real-world agents often need:

1. Database lookup – fetch user profile or inventory.
2. External API – call a weather service or a payment gateway.
3. Long-term memory – persist conversation across sessions.

Adding a database tool

Assume a PostgreSQL table users(id, name, balance). Install asyncpg and add the tool:

import asyncpg
import json

async def get_balance(user_id: int) -> str:
    conn = await asyncpg.connect(os.getenv("DATABASE_URL"))
    row = await conn.fetchrow("SELECT balance FROM users WHERE id = $1", user_id)
    await conn.close()
    if row:
        return f"User {user_id} has a balance of ${row['balance']:.2f}"
    return f"No user found with id {user_id}"

Wrap it for LangChain:

from langchain.tools import BaseTool

class BalanceTool(BaseTool):
    name = "GetBalance"
    description = "Retrieve the account balance for a given user ID."

    async def _run(self, user_id: str) -> str:
        return await get_balance(int(user_id))

Add it to the agent initializer:

balance_tool = BalanceTool()
agent = initialize_agent(
    tools=[calc_tool, balance_tool],
    llm=llm,
    agent_type="zero-shot-react-description",
    memory=memory,
    prompt=prompt,
    verbose=False,
)

Now the agent can decide to call GetBalance when the user asks, “How much money does user 42 have?”

Persistent memory

ConversationBufferMemory lives only in RAM. For multi-instance deployments you need a shared store. I switched to RedisChatMessageHistory:

from langchain.memory import ConversationBufferMemory
from langchain.schema import BaseChatMessageHistory
from langchain.chat_message_histories import RedisChatMessageHistory

redis_history = RedisChatMessageHistory(url="redis://localhost:6379/0", session_id="session_123")
memory = ConversationBufferMemory(chat_memory=redis_history, memory_key="chat_history")

Now any FastAPI replica can read the same conversation history, which is essential when you run behind a load balancer.

When NOT to add memory

If your use-case is stateless (e.g., one-off calculation), persisting history adds latency and cost for no benefit. Also, be mindful of GDPR – storing personal data in Redis requires encryption and proper retention policies.

How should I deploy and scale the agent in production?

Containerization

Create a lightweight Docker image based on python:3.11-slim. Example Dockerfile:

FROM python:3.11-slim

WORKDIR /app
COPY pyproject.toml poetry.lock ./
RUN pip install poetry && poetry config virtualenvs.create false && poetry install --no-dev

COPY . .

EXPOSE 8000
CMD ["uvicorn", "main:app", "--host", "0.0.0.0", "--port", "8000"]

Build and push:

docker build -t yourrepo/langchain-agent:latest .
docker push yourrepo/langchain-agent:latest

Orchestration

I run the service on Kubernetes with a Deployment of three replicas behind an Ingress. The key settings:

apiVersion: apps/v1
kind: Deployment
metadata:
  name: langchain-agent
spec:
  replicas: 3
  selector:
    matchLabels:
      app: langchain-agent
  template:
    metadata:
      labels:
        app: langchain-agent
    spec:
      containers:
        - name: api
          image: yourrepo/langchain-agent:latest
          envFrom:
            - secretRef:
                name: openai-secret
          resources:
            limits:
              cpu: "500m"
              memory: "512Mi"
          ports:
            - containerPort: 8000
          livenessProbe:
            httpGet:
              path: /health
              port: 8000
            initialDelaySeconds: 5
            periodSeconds: 10

Why three replicas?
LangChain calls are I/O bound (network to OpenAI). With three pods you can handle ~30 RPS before the CPU caps out. If you need higher throughput, increase the replica count or move to a larger instance type.

Cost considerations

• LLM usage – each token costs. Enable a caching layer (Redis) for identical prompts. In my last project we saved ~20 % by caching repetitive “What is the weather in London?” queries.
• Thread pool size – each concurrent request spawns a thread for the sync agent. Over-provisioning leads to context-switch thrashing. Tune max_workers to CPU * 2 as a rule of thumb.

Monitoring and alerts

• Prometheus metrics from FastAPI (uvicorn exposes --metrics flag). Track request_latency_seconds, error_total.
• OpenAI usage – export logs to a file, ship to Loki, and alert when daily spend exceeds a threshold.
• Health endpoint – implement /health that pings Redis and runs a cheap LLM prompt like “Say OK”. This catches both DB and API outages.

When to avoid LangChain in production

• Deterministic logic – if the problem can be solved with pure code, skip the LLM. The added latency and cost rarely justify a language-model wrapper.
• Strict latency SLAs – even with a fast model, the network round-trip can be 100 ms+; add a buffer if you need sub-50 ms responses.

FAQ

What model should I use for a production LangChain agent?
Start with gpt-4o-mini or claude-3.5-sonnet for a good cost-performance balance. Reserve the larger gpt-4o for tasks that need richer reasoning.

Can I run LangChain locally without OpenAI?
Yes. LangChain supports Ollama, Llama.cpp, and other self-hosted backends. The API surface is the same, but you’ll need to manage GPU memory and model updates yourself.

How do I prevent the agent from calling a malicious tool?
Define a static whitelist of Tool objects and let the executor raise an error if the model proposes an unknown name. Also, keep tool functions pure and sandbox any external calls.

Is async support coming to LangChain agents?
Async tool execution is already possible via await tool.arun(...), but the high-level initialize_agent still runs synchronously. Until the library fully embraces async, wrap calls in asyncio.to_thread as shown above.

Key Takeaways

• LangChain agents follow a reasoning loop; understand the loop to debug hallucinations.
• Isolate dependencies with Poetry; pin versions to avoid silent breaking changes.
• Wrap sync agents in asyncio.to_thread when exposing them via FastAPI.
• Add memory only when you truly need cross-request context; Redis works well for distributed setups.
• Guard tools with a whitelist and sandbox any external execution.
• Containerize and orchestrate with Kubernetes; three replicas plus a modest thread pool handle typical production loads.
• Monitor spend closely; caching identical prompts can cut token costs dramatically.

If you’re interested in a deeper case study, check out my post on How to build AI agent for stock analysis with FastAPI and the comparison of libraries in Comparing AI Agents Python Library Options for Production.