DEV Community: Ayush Kumar

Redis caching in FastAPI: practical setup and pitfalls

Ayush Kumar — Mon, 03 Aug 2026 07:47:29 +0000

Quick answer:

If you want fast, cheap look-ups for query results, session data, or computed values, add a Redis instance, connect with aioredis, wrap your endpoints in a small decorator, and let the cache expire or be cleared on writes. The whole stack fits in a few dozen lines and costs pennies per month on most cloud providers.

Below I walk through a production-ready implementation of redis caching in fastapi. I’ll show you how to install Redis, wire an async client, build reusable decorators, handle invalidation, spin the service up with Docker, and keep an eye on performance once you’re in the wild. I’ll also point out the places where things tend to break and what I would do differently next time.

Installing and configuring Redis for a FastAPI project

Do I need a separate Redis server?

Yes. Even a single-node instance gives you in-memory speed without adding latency to your API process. For local development you can run the official Docker image; for production you’ll want a managed service or a replicated cluster.

Local Docker setup

# docker-compose.yml
version: "3.9"
services:
  redis:
    image: redis:7-alpine
    ports:
      - "6379:6379"
    command: ["redis-server", "--save", "60", "1", "--loglevel", "warning"]
    restart: unless-stopped

Run docker-compose up -d redis. The container is tiny (≈30 MB) and starts in a second. On a VM or Kubernetes you would expose the service via a ClusterIP or a cloud-managed endpoint.

Production connection string

# .env
REDIS_URL=redis://:myStrongPassword@redis-prod.internal:6379/0

Never hard-code credentials. Keep the URL in a secret manager and inject it at runtime. I’ve been bitten by missing redis:// prefixes when moving from localhost to a managed instance - FastAPI will raise a ValueError at startup.

How do I integrate an async Redis client with aioredis?

Can I use the same client for all requests?

Yes, but you should create a single connection pool at startup and share it via FastAPI’s dependency system. aioredis (now part of redis-py v4) gives you native asyncio support.

# app/redis.py
import os
from redis.asyncio import Redis
from fastapi import FastAPI

def get_redis_pool() -> Redis:
    return Redis.from_url(os.getenv("REDIS_URL"), decode_responses=True)

def add_redis(app: FastAPI) -> None:
    @app.on_event("startup")
    async def startup_redis():
        app.state.redis = get_redis_pool()
        # Test the connection early
        await app.state.redis.ping()

    @app.on_event("shutdown")
    async def shutdown_redis():
        await app.state.redis.close()

In any route you can now do:

from fastapi import Depends, Request

async def redis_dep(request: Request) -> Redis:
    return request.app.state.redis

Trade-offs

Single pool – reduces connection overhead but can become a bottleneck if you saturate the event loop with heavy Redis work. In that case, spawn a second pool for low-latency reads.
Blocking commands – avoid commands that return huge payloads; they block the event loop while the client decompresses data. Use SCAN instead of KEYS in production.

How can I build reusable cache decorators for FastAPI endpoints?

Is a decorator enough for all use cases?

For read-only or idempotent endpoints, a decorator is clean and keeps the view logic tidy. For write-heavy resources you’ll need explicit invalidation.

# app/cache.py
import functools
import json
from typing import Callable, Any
from redis.asyncio import Redis

def redis_cache(key_builder: Callable[..., str],
                ttl: int = 300):
    """Cache the result of an async endpoint.
    key_builder receives the same args/kwargs as the endpoint.
    """
    def decorator(func: Callable):
        @functools.wraps(func)
        async def wrapper(*args, **kwargs):
            request = kwargs.get("request")
            redis: Redis = request.app.state.redis  # type: ignore
            cache_key = key_builder(*args, **kwargs)
            cached = await redis.get(cache_key)
            if cached is not None:
                return json.loads(cached)

            result = await func(*args, **kwargs)
            await redis.set(cache_key, json.dumps(result), ex=ttl)
            return result
        return wrapper
    return decorator

Example usage

# app/routers/items.py
from fastapi import APIRouter, Request
from .cache import redis_cache

router = APIRouter()

def item_key(request: Request, item_id: int):
    return f"item:{item_id}"

@router.get("/items/{item_id}")
@redis_cache(key_builder=item_key, ttl=600)
async def get_item(request: Request, item_id: int):
    # Simulate expensive DB call
    return {"id": item_id, "value": "expensive computation"}

The decorator automatically reuses the pool created earlier, serializes the response with json, and respects the TTL.

When not to use it

Streaming responses – you cannot cache a generator without buffering the whole payload.
Very large payloads – they waste memory and network bandwidth. Consider pagination or a separate blob store.

What are the best cache invalidation and expiration strategies?

How do I keep stale data from leaking?

Time-based expiration (TTL) – simplest. Set a reasonable ttl per endpoint. For data that changes every few minutes, a 60-second TTL is enough.
Write-through invalidation – after you write to the database, delete the related keys.

# app/routers/items.py
@router.put("/items/{item_id}")
async def update_item(request: Request, item_id: int, payload: dict):
    # Update DB here …
    await request.app.state.redis.delete(f"item:{item_id}")
    return {"status": "updated"}

Tag-based eviction – store a set of keys per logical group and delete the whole set on a bulk change.

# add tag on set
await redis.sadd(f"tag:user:{user_id}", cache_key)
# later invalidate all
keys = await redis.smembers(f"tag:user:{user_id}")
if keys:
    await redis.delete(*keys)
    await redis.delete(f"tag:user:{user_id}")

Pitfalls

Forgotten invalidation – I once added a new PUT endpoint and missed the delete call. The cache kept returning the old version for hours. Adding a unit test that checks cache freshness saved me a lot of tickets.
Over-eager expiration – setting TTLs too low defeats the purpose of caching and adds load to Redis. Profile your read/write ratio before deciding.

How do I deploy Redis with Docker and handle environment-specific settings?

Can I keep the same compose file for dev and prod?

Yes, but you should separate the image version and resource limits. In production you may want persistence, ACLs, and monitoring side-cars.

Production-ready compose snippet

# docker-compose.prod.yml
version: "3.9"
services:
  redis:
    image: redis:7-alpine
    command:
      - redis-server
      - /usr/local/etc/redis/redis.conf
    volumes:
      - redis-data:/data
      - ./redis.conf:/usr/local/etc/redis/redis.conf:ro
    environment:
      - REDIS_PASSWORD=${REDIS_PASSWORD}
    ports:
      - "6379:6379"
    deploy:
      resources:
        limits:
          memory: 512M
        reservations:
          memory: 256M
    restart: unless-stopped

volumes:
  redis-data:

redis.conf can enable ACLs:

requirepass ${REDIS_PASSWORD}
maxmemory 256mb
maxmemory-policy allkeys-lru

When deploying to Google Cloud Run you cannot run a side-car, so you’ll point the FastAPI container to a managed Memorystore instance instead. The same REDIS_URL env var works for both setups.

What I’d do differently

Separate config repo – keep redis.conf in a dedicated folder, version it, and inject secrets via the orchestrator. I once edited the file inside the container and lost the change after a redeploy.
Health checks – add a simple redis-cli ping health endpoint in your orchestrator. Without it, a pod could start while Redis was still booting, causing a cascade of connection errors.

How can I monitor and troubleshoot Redis cache performance in production?

What metrics matter most?

Hit-rate, latency, memory usage, and evicted keys. Redis ships with INFO and the Redis Exporter for Prometheus.

Prometheus exporter

# docker-compose.yml (add beside redis)
redis-exporter:
  image: oliver006/redis_exporter
  environment:
    - REDIS_ADDR=redis://:${REDIS_PASSWORD}@redis:6379
  ports:
    - "9121:9121"

Grafana dashboards can then show redis_keyspace_hits_total / (hits + misses) as a percentage. A dip below 80 % usually means your TTL is too short or you’re missing a cache-warming step.

Logging slow commands

Add to redis.conf:

slowlog-log-slower-than 10000   # microseconds
slowlog-max-len 128

You can query SLOWLOG GET from your FastAPI admin endpoint to see which queries are eating time. In one project I discovered a ZRANGEBYSCORE on a 10 M-element sorted set that took 250 ms; the fix was to add a secondary index.

Common failure modes

Symptom	Likely cause	Quick fix
`ConnectionError: Timeout`	Redis container restarted or network partition	Check Docker logs, add `restart: always`, verify health probe
Cache always miss	Wrong key format or missing `decode_responses=True`	Ensure the same key builder is used for set and get
Memory limit reached	`maxmemory` too low, or unbounded keys	Enable LRU eviction, monitor `used_memory_peak`

FAQ

How many Redis nodes do I need for a small FastAPI service?

A single node is fine for low-traffic workloads. Add a replica once you exceed ~100 K requests per second or need high availability.

Can I use Redis as a message broker instead of a cache?

Technically yes, but Pub/Sub semantics differ from a durable queue. For background jobs I prefer RabbitMQ or Google Pub/Sub.

Is it safe to store user-session data in Redis without encryption?

If the data is not personally identifiable, plain storage is okay. For anything sensitive, enable TLS on the Redis endpoint and store only a session token that references server-side data.

What’s the cost difference between managed Memorystore and a self-hosted Docker container?

Managed services start at a few dollars per GB per month and include automatic failover. A self-hosted container on a small VM can be cheaper (< $5) but you’re responsible for backups and scaling.

Key Takeaways

Install Redis as a separate service; use Docker locally and a managed instance in prod.
Create a single async Redis pool at FastAPI startup and inject it via request state.
Wrap read-only endpoints in a lightweight decorator that builds keys, checks the cache, and sets a TTL.
Invalidate on writes with explicit delete calls or tag-based sets for bulk eviction.
Deploy with environment-specific redis.conf, enable ACLs, and set sensible memory limits.
Export metrics to Prometheus, watch hit-rate and latency, and enable the slow-log for troubleshooting.

With these pieces in place, redis caching in fastapi becomes a low-maintenance performance boost that scales from a laptop dev box to a multi-region production fleet. Happy caching!

Kafka Python Tutorial: From Docker to FastAPI

Ayush Kumar — Mon, 03 Aug 2026 07:47:20 +0000

If you need a quick way to get Kafka running locally and talk to it from Python, this kafka python tutorial shows you how to spin up a broker with Docker, install the kafka-python client, produce messages, consume them asynchronously, and hook the flow into a FastAPI endpoint for background processing. By the end you’ll have a reproducible dev environment and a pattern you can push to production with minimal surprises.

How do I spin up a local Kafka broker with Docker?

The simplest way to start experimenting is to run a single-node Kafka cluster in Docker. The official confluentinc/cp-kafka image includes Zookeeper, so you don’t have to manage a separate service.

docker network create kafka-net

docker run -d \
  --name zookeeper \
  --network kafka-net \
  -p 2181:2181 \
  -e ZOOKEEPER_CLIENT_PORT=2181 \
  confluentinc/cp-zookeeper:7.5.0

docker run -d \
  --name kafka \
  --network kafka-net \
  -p 9092:9092 \
  -e KAFKA_BROKER_ID=1 \
  -e KAFKA_ZOOKEEPER_CONNECT=zookeeper:2181 \
  -e KAFKA_ADVERTISED_LISTENERS=PLAINTEXT://localhost:9092 \
  -e KAFKA_OFFSETS_TOPIC_REPLICATION_FACTOR=1 \
  confluentinc/cp-kafka:7.5.0

A few things bite me early:

Port mapping – PLAINTEXT://localhost:9092 works for local dev but will need a proper DNS name in production.
Resource limits – the default container uses ~1 GiB RAM. On a low-spec laptop you’ll see OOM kills unless you set -e KAFKA_HEAP_OPTS="-Xmx512M -Xms512M".
Data persistence – without a volume, every docker rm wipes your topic data. Add -v kafka-data:/var/lib/kafka/data if you need to keep it across restarts.

Verify the broker is alive:

docker exec kafka kafka-topics --bootstrap-server localhost:9092 --list

You should see an empty list, which means the broker is reachable.

How to install and configure the kafka-python client?

For most quick prototypes kafka-python is enough. It’s pure Python, has a small footprint, and works well with FastAPI’s async model when you run the consumer in a background thread.

pip install kafka-python

Configuration is just a dict you pass to the KafkaProducer or KafkaConsumer. In production you’ll pull these values from environment variables or a config service.

KAFKA_CFG = {
    "bootstrap_servers": "localhost:9092",
    "security_protocol": "PLAINTEXT",   # switch to SASL_SSL in prod
    "client_id": "logicloop-producer",
}

A common mistake is to forget the acks setting on the producer. The default is acks=1, which can lose data if the leader crashes. For a safe dev run set:

producer = KafkaProducer(**KAFKA_CFG, acks="all", retries=5)

How to produce messages from a Python application?

Producing is straightforward. Serialize your payload to JSON (or Avro, if you have a schema registry) and send it to a topic.

import json
from kafka import KafkaProducer

producer = KafkaProducer(
    **KAFKA_CFG,
    value_serializer=lambda v: json.dumps(v).encode("utf-8")
)

def send_user_event(user_id: int, action: str):
    payload = {"user_id": user_id, "action": action, "ts": time.time()}
    future = producer.send("user-events", value=payload)
    try:
        record_metadata = future.get(timeout=10)
        print(f"Sent to {record_metadata.topic}:{record_metadata.partition}")
    except Exception as exc:
        print(f"Send failed: {exc}")

# Example usage
send_user_event(42, "login")
producer.flush()

What bites me in production is message ordering. Kafka only guarantees order within a partition. If you need strict ordering per user_id, hash the key to a single partition:

producer = KafkaProducer(
    **KAFKA_CFG,
    key_serializer=str.encode,
    value_serializer=lambda v: json.dumps(v).encode("utf-8")
)

producer.send("user-events", key=str(user_id), value=payload)

Now all events for the same user_id land in the same partition and preserve order.

How to consume messages in Python with async support?

kafka-python does not expose an async API, but you can run the consumer inside an asyncio task that delegates to a thread pool. The pattern works fine for FastAPI background workers.

import asyncio
import json
from kafka import KafkaConsumer
from concurrent.futures import ThreadPoolExecutor

consumer = KafkaConsumer(
    "user-events",
    **KAFKA_CFG,
    group_id="logicloop-group",
    value_deserializer=lambda v: json.loads(v.decode("utf-8")),
    auto_offset_reset="earliest",
)

executor = ThreadPoolExecutor(max_workers=2)

async def consume_loop():
    loop = asyncio.get_running_loop()
    while True:
        # poll blocks, so we run it in a thread
        msgs = await loop.run_in_executor(executor, consumer.poll, 1.0)
        for tp, records in msgs.items():
            for record in records:
                await handle_event(record.value)

async def handle_event(event: dict):
    # Simulate async DB write
    await asyncio.sleep(0.01)
    print(f"Processed {event}")

# Run the loop in a separate thread when the app starts
def start_consumer_background():
    asyncio.run(consume_loop())

If you prefer a truly async client, swap kafka-python for aiokafka. The API is almost identical, but you must install the extra package and adjust the startup code. I keep the kafka-python version for simplicity because it plays nicely with the existing codebase and the extra thread overhead is negligible for low-throughput services.

When I first tried to run the consumer directly in the FastAPI event loop, the app hung on startup. The fix was to launch the consumer in a separate daemon thread (or use lifespan events). See the FastAPI async pitfalls in the internal link Fixing SQLAlchemy MissingGreenlet Error in FastAPI (Async Explained) for more context on mixing sync and async code.

How can I integrate Kafka with FastAPI for background processing?

FastAPI gives you two natural hooks: startup events to launch the consumer and background tasks to fire-and-forget work that originates from HTTP requests.

from fastapi import FastAPI, BackgroundTasks

app = FastAPI()

@app.on_event("startup")
async def start_kafka_consumer():
    loop = asyncio.get_event_loop()
    loop.create_task(consume_loop())   # consume_loop from previous section

@app.post("/events/")
async def publish_event(user_id: int, action: str, background: BackgroundTasks):
    send_user_event(user_id, action)   # sync producer, cheap call
    # Optionally schedule extra processing
    background.add_task(log_audit, user_id, action)
    return {"status": "queued"}

def log_audit(user_id: int, action: str):
    # This runs in a thread pool managed by FastAPI
    time.sleep(0.1)  # simulate I/O
    print(f"Audit log for {user_id}:{action}")

A few production-grade concerns:

Graceful shutdown – FastAPI’s shutdown event should close the consumer and the thread pool to avoid “dangling thread” warnings.
Back-pressure – If the consumer can’t keep up, messages accumulate in the broker. Tune max_poll_records and consider scaling the consumer group horizontally.
Observability – Export consumer lag metrics to Prometheus. The kafka-python client exposes consumer.position(tp) and consumer.end_offsets([tp]) which you can compare.

When I first deployed this pattern to Google Cloud Run, the container was killed after 15 minutes of idle time, causing the consumer to miss messages. The fix was to add a tiny “keep-alive” endpoint that Cloud Run pings, and to set the --timeout flag higher. The full deployment story is covered in Serverless Python: Deploying FastAPI to Google Cloud Run with Docker.

Common pitfalls and best-practice tips

Pitfall	Why it hurts	Remedy
Using the default `acks=1` in production	Possible data loss on leader failure	Set `acks="all"` and enable idempotent producer (`enable_idempotence=True`)
Forgetting to commit offsets manually	Consumer may reprocess the same batch after restart	Use `consumer.commit()` after successful handling, or enable `enable_auto_commit=False` and manage commits yourself
Running the consumer in the same thread as FastAPI	Event loop blocks, endpoint latency spikes	Run the consumer in a separate daemon thread or use `aiokafka`
Not configuring retries on the producer	Transient network blips cause message drop	Set `retries` and `retry_backoff_ms`
Over-committing partitions to a single consumer	Limits scalability	Keep the number of partitions ≥ number of consumer instances
Ignoring broker resource limits	Out-of-memory kills in Docker	Allocate appropriate heap (`KAFKA_HEAP_OPTS`) and monitor container memory

When you move beyond a single node, you’ll need to think about replication factor, multi-region latency, and security (SASL/SCRAM, TLS). Those concerns are out of scope for this tutorial but worth planning early.

FAQ

What’s the difference between kafka-python and aiokafka?

kafka-python is a synchronous client that works well with a thread-based approach. aiokafka is built on asyncio and offers true non-blocking APIs, but adds an extra dependency and a slightly steeper learning curve.

Do I need Zookeeper for a production Kafka cluster?

Modern Kafka versions can run without an external Zookeeper when you enable KRaft mode. The Docker images used in this tutorial still rely on Zookeeper for simplicity, but for new clusters consider KRaft to reduce operational overhead.

How many partitions should I create for a topic?

Start with a number that matches your expected consumer parallelism (e.g., 3-6 partitions for a small service). You can increase partitions later, but you cannot decrease them without recreating the topic.

Can I use this setup on Windows?

Docker Desktop runs Linux containers on Windows, so the broker part works the same. The only Windows-specific issue is the path handling in volume mounts; use absolute Unix-style paths inside the container.

Key Takeaways

Docker lets you spin up a single-node Kafka broker in seconds; remember to persist data if you need it across restarts.
kafka-python is a lightweight client; configure acks, retries, and proper serialization to avoid data loss.
Produce messages with a key if you need ordering per entity.
Consume asynchronously by delegating the blocking poll to a thread pool; for pure async workloads consider aiokafka.
FastAPI startup/shutdown events are the right place to launch and stop a Kafka consumer; background tasks handle fire-and-forget work.
Watch for common pitfalls: offset management, back-pressure, resource limits, and security settings.

With these pieces in place you have a solid foundation for building reliable, event-driven services in Python. The next step is to push the Docker image to your CI/CD pipeline - see the guide on Automating Production: A CI/CD Pipeline for Google Cloud Run with GitHub Actions for a production-ready workflow.

Fixing AI Generated Code Quality Issues in Production

Ayush Kumar — Fri, 31 Jul 2026 10:31:32 +0000

When AI-generated code quality issues show up in production, the quickest fix is to stop the broken code from reaching users and put a safety net around the generation step. In short: add tests, lint, and a CI gate, then treat every AI snippet like a third-party library you would audit before import.

I’ve been bitten by this more than once. A single line of code that a tool like Cursor or Claude wrote crashed my FastAPI service under load, and the cost was a missed SLA and a frantic hot-fix sprint. Below I walk through the exact failure patterns I’ve seen, the tooling that catches them early, and the process tweaks that keep the pipeline clean. If you’re at the point where “it works locally but blows up in prod” feels like a mantra, this is for you.

What are the most common AI generated code quality pitfalls?

The first thing to know is that AI-generated code isn’t magically perfect. It often repeats three classic mistakes:

Missing edge-case handling – The snippet assumes happy-path inputs. In my last project, an AI-written validation function accepted any string, forgetting to strip whitespace. A single malformed request caused a ValueError that bubbled up to the FastAPI exception handler and returned a 500.
Incorrect async/sync mix – FastAPI is async-first, yet the model sometimes drops a blocking requests.get into an async endpoint. The result? The event loop stalls, latency spikes, and the autoscaler thinks the service is overloaded.
Undocumented side effects – The generated code may open files, create temp directories, or mutate global state without a comment. In a microservice that runs many concurrent requests, that leads to race conditions and flaky tests.

These pitfalls manifest as ai generated code quality issues that are easy to miss in a quick copy-paste but explode under real traffic.

How can I test AI-generated snippets automatically?

The answer is: treat every snippet as a unit that must pass the same test suite as the rest of your code. Start with a tiny harness that runs the snippet in isolation.

# test_generated_snippet.py
import pytest
from importlib import util
from pathlib import Path

def load_snippet(path: Path):
    spec = util.spec_from_file_location("generated", path)
    module = util.module_from_spec(spec)
    spec.loader.exec_module(module)
    return module

@pytest.fixture
def snippet():
    return load_snippet(Path("generated/validation.py"))

def test_input_validation(snippet):
    # Assume the AI created a function called `validate_user_input`
    assert snippet.validate_user_input("valid123") is True
    with pytest.raises(ValueError):
        snippet.validate_user_input("  bad  ")

Run this with pytest -q. If the snippet fails, you know the problem before it reaches CI. The same pattern works for async code - use pytest-asyncio and assert that no blocking calls are made.

Trade-off: Writing a harness for every snippet adds a few minutes of developer time, but it saves hours of debugging in production. If you generate dozens of snippets per week, automate harness creation with a small script that reads a comment header from the AI output (e.g., # test:validation).

Which static analysis and linting tools catch the most issues?

Static analysis is the cheapest guardrail. Here are the tools I rely on for ai generated code quality issues:

Tool	What it catches	Why it matters
ruff	Unused imports, undefined names, line length, async-sync mismatches	Fast, integrates with pre-commit
mypy	Type mismatches, missing return annotations	Forces the AI to be explicit
bandit	Security-oriented patterns (e.g., `eval`, insecure HTTP)	Prevents accidental vulnerabilities
pylint (configured with `disable=missing-docstring`)	Code smells, duplicate code, too complex functions	Highlights refactor opportunities

A minimal .pre-commit-config.yaml that runs these on every commit looks like this:

repos:
  - repo: https://github.com/charliermarsh/ruff-pre-commit
    rev: v0.5.0
    hooks:
      - id: ruff
  - repo: https://github.com/pre-commit/mirrors-mypy
    rev: v1.10.0
    hooks:
      - id: mypy
        additional_dependencies: [types-requests]
  - repo: https://github.com/PyCQA/bandit
    rev: 1.7.9
    hooks:
      - id: bandit
        args: ["-r", "generated/"]

Run pre-commit install and you’ll get instant feedback when an AI snippet violates any rule. In practice, I caught a missing await call that would have blocked the entire FastAPI worker within seconds of the snippet being saved.

When not to use: If you generate code in a language other than Python, swap in the appropriate linters (e.g., eslint for JavaScript). The principle stays the same.

How do I integrate quality gates for AI code into CI/CD?

The answer: add a dedicated “AI-generated” stage to your pipeline that runs the same linters and tests, and make it a required check before merge.

A typical GitHub Actions workflow:

name: CI

on: [push, pull_request]

jobs:
  lint-and-test:
    runs-on: ubuntu-latest
    steps:
      - uses: actions/checkout@v4
      - name: Set up Python
        uses: actions/setup-python@v5
        with:
          python-version: "3.12"
      - name: Install deps
        run: pip install -r requirements.txt
      - name: Run linters
        run: |
          ruff .
          mypy .
          bandit -r generated/
      - name: Run tests
        run: pytest -q

If any step fails, the PR cannot be merged. The extra cost is a few minutes of CI time, but it prevents the far more expensive rollback of a production incident.

Integration tip: Tag AI-generated files with a comment header, e.g., # AI_GENERATED. Then configure your linter to only scan those paths, keeping the rest of the repo fast.

What are best practices for reviewing and refactoring AI-generated code?

Even with tests and linters, a human eye is still valuable. Here’s my checklist:

Read the intent comment – Most generators let you prepend a description. Verify that the comment matches the implementation.
Check for hidden state – Look for global variables, file writes, or network calls. Wrap them in a context manager if they’re needed.
Add type hints – If the AI omitted them, add explicit TypedDict or pydantic models. This forces downstream code to respect the contract.
Isolate the snippet – Move it to its own module under generated/ and expose a clean public API. That makes future refactors easier.
Document edge cases – Write a short docstring that lists the inputs the function does not handle. This prevents the same mistake from being re-generated later.

When you refactor, keep the original snippet versioned. I store the raw AI output in a separate raw/ folder; that way I can compare before and after with git diff. It also helps when you need to audit who generated what, a common request from security teams.

If you need a concrete example, see how I turned an AI-written stock-analysis endpoint into a well-typed FastAPI router in the post How to build AI agent for stock analysis with FastAPI. The transformation reduced latency by 30 % and eliminated a hidden requests call that was blocking the event loop.

FAQ

Why does my AI-generated async function still block the event loop?

Because the snippet likely contains a synchronous library call (e.g., requests.get). Replace it with an async client like httpx or run the call in a thread pool with anyio.to_thread.run_sync.

Can I trust linting alone to catch security bugs in AI code?

No. Linters flag risky patterns, but a manual security review or a tool like bandit is still required for deeper analysis.

Do I need to write unit tests for every single AI snippet?

Ideally yes, but a pragmatic approach is to prioritize snippets that touch I/O, authentication, or external APIs. Those are the ones that cause the biggest production pain.

What if the AI keeps generating the same bug after I fix it?

Add the fix to the prompt you feed the model, or use a post-generation script that applies a template patch. Over time the model learns from the corrected examples.

Key Takeaways

Identify the failure: missing edge-case handling, async/sync mismatches, and hidden side effects are the most common AI generated code quality issues.
Test early: a small pytest harness for each snippet catches bugs before they enter CI.
Lint aggressively: ruff, mypy, and bandit together spot the majority of problems.
Gate the pipeline: make linting and testing a required CI stage to stop bad code from merging.
Review thoughtfully: isolate snippets, add type hints, and document edge cases to make future maintenance painless.

If you’ve hit a wall and need someone to step in, I’m taking on a few hands-on contracts right now. Feel free to reach out through the hire page and we can tackle those stubborn AI generated code quality issues together.

Fixing an AI Generated Code Vulnerabilities Report

Ayush Kumar — Fri, 31 Jul 2026 09:29:21 +0000

I’ve been there: you let an LLM write a FastAPI endpoint, the app ships, and a security scan flags a handful of issues you never saw coming. The good news is you don’t have to chase down each problem manually. An ai generated code vulnerabilities report tells you exactly what to fix, and with the right tooling you can make that report part of every pull request. Below I walk through why AI-generated snippets are prone to bugs, how to set up static analysis tools, automate scans in CI/CD, read the report, and apply best-practice mitigations.

Why does AI-generated code often contain security flaws?

The answer is simple: language models predict the next token, not the next safe token. They have seen millions of open-source projects, many of which contain outdated patterns, hard-coded secrets, or insecure defaults. When you ask a model to “write a login route with JWT”, it will often:

Use a static secret key (SECRET = "mysecret").
Skip input validation (username = request.json["user"]).
Trust user-provided data for file paths.

Those patterns compile and run, but they open the door to injection, credential leakage, and privilege escalation. In production I saw a bot-generated CSV import that used os.system on the supplied filename – a classic command injection that slipped through code review because the snippet was only a few lines.

Static analysis tools shine here because they look for the effect of the code, not the intent behind it. They can flag hard-coded secrets, unsafe deserialization, missing CSRF protection, and more. The key is to treat the output of the LLM as any other third-party dependency: run it through a scanner before it touches your codebase.

How do I set up static analysis tools for AI-generated snippets?

I use a mix of Bandit, Semgrep, and SonarQube. Bandit is quick for Python-specific issues, Semgrep lets me write custom rules that target the patterns LLMs love, and SonarQube provides a dashboard for the whole team.

1. Install Bandit

pip install bandit

Create a simple wrapper that runs Bandit only on the generated/ folder where you store AI-produced files:

# run_bandit.sh
#!/usr/bin/env bash
set -e
bandit -r generated/ -lll -f json -o bandit_report.json

The -lll flag treats low-severity findings as failures, which is useful when you want a clean bill of health before merging.

2. Add Semgrep with custom rules

First, install Semgrep:

pip install semgrep

Then create a rule file semgrep_rules.yml that catches common LLM pitfalls:

rules:
  - id: hardcoded-secret
    patterns:
      - pattern: $VAR = "$SECRET"
    message: "Avoid hard‑coded secrets in generated code."
    severity: HIGH
    languages: [python]
  - id: unsafe-os-system
    pattern: os.system(...)
    message: "os.system can lead to command injection."
    severity: HIGH
    languages: [python]

Run it with:

semgrep --config semgrep_rules.yml generated/ --json > semgrep_report.json

3. Hook SonarQube into the mix

If you already have SonarQube for the rest of the repo, just add the generated/ directory to the analysis scope in sonar-project.properties:

sonar.sources=app,generated
sonar.exclusions=tests/**

When you trigger a SonarQube scan, the UI will aggregate findings from both your hand-written code and the AI-generated snippets, giving you a single ai generated code vulnerabilities report.

How can I automate vulnerability scanning in my CI/CD pipeline?

Automation is the only way to keep the feedback loop short. Below is a GitHub Actions workflow that runs Bandit, Semgrep, and SonarQube on every PR. Adjust the steps if you use GitLab or CircleCI – the concepts are the same.

name: Scan AI‑Generated Code

on:
  pull_request:
    paths:
      - 'generated/**'

jobs:
  security-scan:
    runs-on: ubuntu-latest
    steps:
      - uses: actions/checkout@v3

      # Set up Python
      - name: Set up Python
        uses: actions/setup-python@v4
        with:
          python-version: '3.11'

      # Install tools
      - name: Install scanners
        run: |
          pip install bandit semgrep
          # SonarScanner CLI
          curl -sSLo sonar-scanner.zip https://binaries.sonarsource.com/Distribution/sonar-scanner-cli/sonar-scanner-cli-5.0.1.3009-linux.zip
          unzip sonar-scanner.zip -d $HOME
          echo "$HOME/sonar-scanner-5.0.1.3009-linux/bin" >> $GITHUB_PATH

      # Run Bandit
      - name: Run Bandit
        run: ./run_bandit.sh

      # Run Semgrep
      - name: Run Semgrep
        run: semgrep --config semgrep_rules.yml generated/ --json > semgrep_report.json

      # Run SonarQube
      - name: SonarQube Scan
        env:
          SONAR_TOKEN: ${{ secrets.SONAR_TOKEN }}
        run: |
          sonar-scanner \
            -Dsonar.projectKey=my-fastapi-app \
            -Dsonar.sources=app,generated \
            -Dsonar.host.url=https://sonar.mycompany.com \
            -Dsonar.login=$SONAR_TOKEN

      # Upload reports as artifacts
      - name: Upload reports
        uses: actions/upload-artifact@v3
        with:
          name: security-reports
          path: |
            bandit_report.json
            semgrep_report.json

The workflow fails the PR if any scanner produces a HIGH severity finding. That way the ai generated code vulnerabilities report becomes a gate, not an after-the-fact checklist.

How do I interpret and triage the generated vulnerability report?

When the pipeline finishes you’ll have three JSON files. Here’s a quick way to turn them into a readable markdown table that you can paste into the PR comment.

import json
from pathlib import Path

def load(path):
    return json.load(Path(path).open())

bandit = load('bandit_report.json')
semgrep = load('semgrep_report.json')
# SonarQube API can be queried for issues; for brevity we assume a local JSON export
sonar = load('sonarqube_issues.json')

def format_issue(issue, source):
    return f"| {source} | {issue['filename']} | {issue.get('line_number', '-') } | {issue['severity']} | {issue['issue_text']} |"

rows = []
for i in bandit['results']:
    rows.append(format_issue(i, 'Bandit'))
for i in semgrep:
    rows.append(format_issue(i, 'Semgrep'))
for i in sonar['issues']:
    rows.append(format_issue(i, 'SonarQube'))

markdown = "\n".join([
    "## AI Generated Code Vulnerabilities Report",
    "| Tool | File | Line | Severity | Description |",
    "|------|------|------|----------|-------------|",
    *rows
])
Path('vuln_report.md').write_text(markdown)
print("Report written to vuln_report.md")

A few triage tips:

Prioritize HIGH severity – they usually map to credential exposure or remote code execution.
Group by file – if a single generated file has three issues, consider rewriting it manually.
Check for duplicates – Bandit and SonarQube may flag the same hard-coded secret; keep only the highest severity entry.
Add remediation notes – in the PR comment, suggest concrete fixes (e.g., “replace static key with os.getenv('JWT_SECRET')”).

When you see the same pattern across many snippets, it’s a signal to adjust your prompt or post-process step. For example, after spotting repeated os.system usage, I added a prompt clause: “Never use os.system; use subprocess.run with a list of arguments”.

What best practices can I follow to mitigate risks in AI-generated code?

Never merge raw AI output – always run it through a linter and a scanner first.
Store generated code in a dedicated directory – makes scoping easier and keeps human-written code separate.
Enforce secret management – replace any literal that looks like a key or password with an environment variable lookup. Tools like python-dotenv help keep defaults empty.
Add unit tests immediately – a failing test catches logic errors that static analysis may miss. I usually generate a skeleton test with the same LLM, then flesh it out manually.
Version-control the prompts – if a specific prompt leads to insecure code, lock it down or add a comment that warns future users.
Run dependency-checkers – AI may suggest outdated libraries. Use pip-audit or safety to verify that the generated requirements.txt is safe.
Consider a “review bot” – a small FastAPI endpoint that receives generated files, runs the scanners, and returns the JSON report. See my post on AI agent python ollama: Build, Test, Deploy with FastAPI for a pattern you can adapt.

When you combine these practices with the automated pipeline above, the ai generated code vulnerabilities report becomes a routine health check rather than a surprise after a production incident.

FAQ

What if the scanners miss a vulnerability?

Static analysis is not a silver bullet. Complement it with runtime monitoring (e.g., Snyk IaC scans, OWASP ZAP for endpoints) and regular pen-testing.

Can I rely on Bandit alone?

Bandit is great for common Python pitfalls, but it doesn’t understand templating or configuration files. Adding Semgrep or SonarQube covers those gaps.

Is it safe to use AI for security-critical code?

Only if you treat the output as untrusted. Run it through the same review, testing, and scanning pipeline you would any third-party library.

How much does this add to CI time?

In my experience the three scanners together add ~30 seconds per PR. The cost is negligible compared to a production breach.

Key Takeaways

AI-generated code often repeats insecure patterns; treat it as third-party code.
Set up Bandit, Semgrep, and SonarQube to create a comprehensive ai generated code vulnerabilities report.
Automate the scans in CI/CD so every pull request is gated by security checks.
Use a small script to turn JSON findings into a readable markdown table for quick triage.
Apply best practices: separate directories, secret management, immediate unit tests, and prompt versioning.
When you need a hand setting this up for your FastAPI service, feel free to reach out at /hire/. I’m happy to help you tighten the loop between AI generation and production security.

How to build AI agent for stock analysis with FastAPI

Ayush Kumar — Sun, 26 Jul 2026 07:26:59 +0000

I’ll show you how to build an AI agent for stock analysis that runs end-to-end in production. The recipe is simple: pull reliable market data, stitch together a LangChain (or CrewAI) workflow, let an LLM write the analysis, add memory for portfolio context, backtest the whole thing, and finally serve it behind FastAPI. Below is a practical walk-through with the code I use in my own trading bots, plus the pitfalls that kept me up at night.

Which market data APIs should I use to build an AI agent for stock analysis?

The first thing that breaks most prototypes is bad data. You need a source that offers near-real-time quotes, historical OHLCV, and a stable free tier if you’re experimenting.

Provider	Real-time?	Historical depth	Free tier	Typical cost
Alpha Vantage	Delayed (15 min)	20 years	5 req/min	$50/mo for premium
IEX Cloud	Real-time (US)	30 years	500 req/day	$9/mo for paid tier
Yahoo Finance (yfinance)	Delayed	25 years	Free (scrape)	N/A
Polygon.io	Real-time + crypto	30 years	5 req/min	$199/mo

For a production-grade agent I combine a cheap real-time feed (IEX) with a free historical dump (yfinance). The real-time feed drives the decision loop; the historical data fuels backtesting and feature engineering.

import yfinance as yf
import requests
import os
from datetime import datetime, timedelta

IEX_TOKEN = os.getenv("IEX_TOKEN")
BASE_URL = "https://cloud.iexapis.com/stable"

def get_realtime_price(symbol: str) -> float:
    url = f"{BASE_URL}/stock/{symbol}/quote?token={IEX_TOKEN}"
    resp = requests.get(url, timeout=5)
    resp.raise_for_status()
    return resp.json()["latestPrice"]

def get_historical(symbol: str, start: str, end: str):
    df = yf.download(symbol, start=start, end=end, progress=False)
    df.reset_index(inplace=True)
    return df

What can go wrong?

Rate limits – IEX will throttle you after a few hundred calls. Cache the last price in Redis and only hit the API when the cache expires (e.g., 60 seconds).

Data mismatches – IEX and yfinance use different ticker conventions; always normalise symbols before you store them.

How do I design the agent architecture with LangChain or CrewAI?

Both LangChain and CrewAI give you a composable graph of LLM calls, tools, and memory. I started with LangChain because its tooling around agents is mature, then migrated a few modules to CrewAI for better parallel tool execution.

A minimal architecture looks like this:

Input layer – FastAPI endpoint receives a ticker list.
Orchestrator – LangChain AgentExecutor decides whether to call a price tool, a news summariser, or the analysis chain.
Tool layer – Python functions wrapped as Tools (price fetch, technical indicator, sentiment).
LLM chain – Prompt template that asks the model to generate a “buy / hold / sell” recommendation with confidence.
Memory – Redis vector store that stores the last N decisions and portfolio exposure.

from langchain.agents import initialize_agent, Tool
from langchain.llms import OpenAI
from langchain.prompts import PromptTemplate
import redis
import json

# Redis for short‑term memory
redis_client = redis.Redis(host="redis", port=6379, db=0)

def price_tool(symbol: str) -> str:
    price = get_realtime_price(symbol)
    return f"The current price of {symbol} is ${price:.2f}"

def sentiment_tool(symbol: str) -> str:
    # Placeholder: call a news API and summarise with LLM
    return f"Sentiment for {symbol} is neutral."

price = Tool(
    name="price",
    func=price_tool,
    description="Get the latest market price for a ticker."
)

sentiment = Tool(
    name="sentiment",
    func=sentiment_tool,
    description="Return a short sentiment summary for a ticker."
)

prompt = PromptTemplate(
    input_variables=["price", "sentiment", "history"],
    template=(
        "You are a quantitative analyst. Given the price: {price}, "
        "sentiment: {sentiment}, and the last 5 decisions: {history}, "
        "output a JSON with fields: decision (buy/hold/sell), confidence (0‑1), "
        "and a one‑sentence rationale."
    )
)

llm = OpenAI(model="gpt-4o-mini", temperature=0.2)
agent = initialize_agent(
    tools=[price, sentiment],
    llm=llm,
    agent_type="zero-shot-react-description",
    verbose=True,
)

def run_analysis(symbol: str):
    # Pull recent decisions from Redis
    past = redis_client.lrange(f"history:{symbol}", -5, -1)
    past_json = [json.loads(x) for x in past] if past else []
    history_str = ", ".join(d["decision"] for d in past_json)

    # Let the agent decide which tools to call
    result = agent.run(
        {
            "price": price_tool(symbol),
            "sentiment": sentiment_tool(symbol),
            "history": history_str,
        }
    )
    decision = json.loads(result)
    # Store decision for future memory
    redis_client.rpush(f"history:{symbol}", json.dumps(decision))
    redis_client.ltrim(f"history:{symbol}", -20, -1)  # keep last 20
    return decision

When not to use this?

If you only need a static statistical model (e.g., a mean-reversion rule), the overhead of an LLM agent is wasted CPU and cost. Also, LangChain’s AgentExecutor adds latency (~300 ms) per call – not ideal for high-frequency strategies.

Failure modes –

Tool hallucination: The LLM may request a tool that isn’t registered, causing a runtime error. Keep the verbose flag on while debugging.

State drift: Forgetting to prune Redis history leads to stale context influencing new decisions.

How can I add memory and tool use for portfolio management?

A stock-analysis agent is only useful if it knows what you already own. I store two kinds of memory:

Short-term – recent recommendations and positions (Redis).
Long-term – aggregated performance metrics (PostgreSQL).

The short-term store is already in the code above. For portfolio exposure I expose a tiny FastAPI route that reads the Redis cache and writes to a PostgreSQL table every hour.

# fastapi_app.py
from fastapi import FastAPI, HTTPException
from pydantic import BaseModel
import asyncpg
import aioredis

app = FastAPI()
redis = aioredis.from_url("redis://redis")
pg_pool = None

class Order(BaseModel):
    symbol: str
    action: str  # buy or sell
    qty: int

@app.on_event("startup")
async def startup():
    global pg_pool
    pg_pool = await asyncpg.create_pool(dsn=os.getenv("DATABASE_URL"))

@app.post("/order")
async def place_order(order: Order):
    # naive simulation of a broker call
    if order.action not in {"buy", "sell"}:
        raise HTTPException(400, "Invalid action")
    # Record in Postgres
    async with pg_pool.acquire() as conn:
        await conn.execute(
            "INSERT INTO trades (symbol, action, qty, ts) VALUES ($1,$2,$3,now())",
            order.symbol, order.action, order.qty,
        )
    # Update memory
    await redis.rpush(f"history:{order.symbol}", json.dumps({
        "decision": order.action,
        "confidence": 1.0,
        "timestamp": datetime.utcnow().isoformat()
    }))
    return {"status": "ok"}

The agent can now query its own position before recommending a trade. Add a simple tool:

def position_tool(symbol: str) -> str:
    # Summarise current net position from Postgres
    async def fetch():
        async with pg_pool.acquire() as conn:
            row = await conn.fetchrow(
                "SELECT SUM(CASE WHEN action='buy' THEN qty ELSE -qty END) AS net "
                "FROM trades WHERE symbol=$1", symbol
            )
            return row["net"] or 0
    net = asyncio.run(fetch())
    return f"Current net position for {symbol} is {net} shares."

Add it to the tools list and the LLM will automatically request it when needed.

Trade-offs –

Consistency: If the PostgreSQL write fails, the Redis cache and DB diverge. Wrap both in a transaction or use an outbox pattern.

Cost: Each LLM call costs a few cents; adding extra tools multiplies the number of calls per analysis. Keep prompts tight.

How do I backtest and deploy the stock analysis agent?

Backtesting proves whether the agent’s logic adds value before you risk real capital. I use backtrader because it integrates nicely with pandas data frames from yfinance.

import backtrader as bt
import pandas as pd

class LLMStrategy(bt.Strategy):
    params = dict(symbol="AAPL", cash=100_000)

    def __init__(self):
        self.dataclose = self.datas[0].close
        self.order = None

    def next(self):
        if self.order:
            return  # wait for order to fill

        decision = run_analysis(self.p.symbol)  # sync call for demo
        if decision["decision"] == "buy" and not self.position:
            self.order = self.buy(size=10)
        elif decision["decision"] == "sell" and self.position:
            self.order = self.sell(size=self.position.size)

# Load data
df = get_historical("AAPL", "2020-01-01", "2023-01-01")
data = bt.feeds.PandasData(dataname=df)

cerebro = bt.Cerebro()
cerebro.addstrategy(LLMStrategy, symbol="AAPL")
cerebro.adddata(data)
cerebro.broker.setcash(100_000)
cerebro.run()
print(f"Final portfolio value: ${cerebro.broker.getvalue():,.2f}")

Production notes

Speed – The backtest calls the LLM for every bar, which is far too slow. Replace the LLM with a deterministic rule derived from the backtest results (e.g., “if confidence > 0.8 and price < MA50, buy”).
Cost – Running a full-year backtest for 10 symbols can cost $30 in API usage. Cache the LLM outputs per day.
Deployment – Containerise the FastAPI service with Docker, use uvicorn --workers 4 behind an Nginx reverse proxy, and schedule a Celery beat job that triggers run_analysis nightly.

# Dockerfile
FROM python:3.11-slim
WORKDIR /app
COPY requirements.txt .
RUN pip install -r requirements.txt
COPY . .
EXPOSE 8000
CMD ["uvicorn", "fastapi_app:app", "--host", "0.0.0.0", "--port", "8000", "--workers", "4"]

Deploy to a Kubernetes cluster with a HorizontalPodAutoscaler that scales on CPU. Keep the Redis and Postgres pods in the same namespace for low latency.

When to skip this pipeline – If you only need a one-off signal for a single ticker, the overhead of Docker, Kubernetes, and backtesting is overkill. A simple script run locally is enough.

FAQ

What’s the cheapest way to get real-time market data for a hobby project?

Use the free tier of IEX Cloud (500 req/day) combined with yfinance for historical data. Cache results aggressively to stay inside the limit.

Can I replace the LLM with a rule-based system?

Yes. If latency or cost is a concern, train a lightweight classifier on the LLM’s historical outputs and use that model in production.

How often should I retrain or update the prompt?

At least monthly, or whenever you notice a drift in confidence scores. Prompt engineering is an ongoing maintenance task.

Do I need a GPU for this stack?

Not for inference with OpenAI’s hosted models. If you run a local model (e.g., Llama-3) you’ll need a GPU, but that adds hardware cost and complexity.

Key Takeaways

Choose a reliable market-data API and cache aggressively; rate limits bite fast.
LangChain (or CrewAI) gives you a clean tool-based agent, but each extra tool adds latency and cost.
Store short-term decisions in Redis and long-term performance in Postgres to keep the agent context-aware.
Backtest with a deterministic surrogate of the LLM to avoid prohibitive API bills.
Deploy the whole stack with FastAPI, Docker, and Kubernetes; monitor Redis, Postgres, and LLM usage to catch failures early.

RAG evaluation using ragas: a practical guide

Ayush Kumar — Sun, 26 Jul 2026 07:26:38 +0000

Why evaluate RAG pipelines and which quality dimensions matter?

If you need to know whether your Retrieval-Augmented Generation (RAG) system is actually helping users, rag evaluation using ragas is the fastest way to get quantitative answers.

A RAG pipeline can break in three invisible ways: the retriever returns irrelevant chunks, the generator hallucinates, or the combination yields low factual consistency. Those failures show up as low scores on relevance, faithfulness, and answer correctness – the core dimensions RAGAS measures.

In production I’ve seen latency budgets explode because developers kept adding more context without checking if it improved the answer. The result? Higher cost, same or worse quality. Measuring early saves you from that waste.

What does RAGAS actually measure?

RAGAS (Retrieval Augmented Generation Assessment Suite) ships with a handful of metrics that map directly to the dimensions above:

Metric	What it captures	Typical range
retrieval_precision	Fraction of retrieved passages that contain the ground-truth answer	0-1
retrieval_recall	Coverage of all ground-truth evidence in the retrieved set	0-1
answer_faithfulness	How much the generated answer stays true to the retrieved passages (using entailment models)	0-1
answer_relevance	Overlap between answer content and the query intent (BLEU, ROUGE, or embedding similarity)	0-1
answer_correctness	Binary check against a gold answer (exact match or fuzzy)	0-1
context_redundancy	Redundancy among retrieved chunks – high redundancy hurts token budget	0-1 (lower is better)

RAGAS also provides a combined score that weights these dimensions according to a user-defined schema. The suite is model-agnostic; you can plug in any LLM or embedding model you already run in production.

How do I install and configure RAGAS in a Python project?

The library is on PyPI and works with Python 3.9+. In a typical FastAPI repo I keep it in the requirements.txt alongside openai, sentence-transformers, and torch.

# In your virtualenv or Dockerfile
pip install ragas[all]  # pulls in optional dependencies like transformers

Configuration is minimal. You need two things:

A retriever function that returns a list of passages given a query.
A generator function that produces an answer from the query and the retrieved passages.

RAGAS expects these functions to accept and return plain Python types, not FastAPI request objects. Here’s a tiny example using sentence-transformers for retrieval and openai for generation:

from typing import List, Tuple
from ragas import evaluate
from sentence_transformers import CrossEncoder, SentenceTransformer
import openai

# 1️⃣ Retriever – dense similarity search
embedder = SentenceTransformer("all-MiniLM-L6-v2")
cross_encoder = CrossEncoder("cross-encoder/ms-marco-MiniLM-L-12-v2")

def retrieve(query: str, docs: List[str], k: int = 5) -> List[str]:
    query_emb = embedder.encode([query])
    doc_embs = embedder.encode(docs)
    scores = cross_encoder.predict([(query, d) for d in docs])
    top_k = sorted(zip(scores, docs), reverse=True)[:k]
    return [doc for _, doc in top_k]

# 2️⃣ Generator – OpenAI ChatCompletion
def generate(query: str, context: List[str]) -> str:
    prompt = (
        "Answer the question using ONLY the following context.\n\n"
        f"Context:\n{''.join(context)}\n\nQuestion: {query}\nAnswer:"
    )
    resp = openai.ChatCompletion.create(
        model="gpt-3.5-turbo",
        messages=[{"role": "user", "content": prompt}],
        temperature=0.0,
    )
    return resp.choices[0].message["content"].strip()

Now you can feed a dataset of (query, ground_truth, documents) tuples to RAGAS:

from ragas import Dataset

# Example dataset entry
samples = [
    {
        "question": "What is the capital of France?",
        "answer": "Paris",
        "contexts": [
            "Paris is the capital city of France, known for the Eiffel Tower.",
            "France's largest city is Paris, which hosts many museums.",
            # ... more docs ...
        ],
    },
    # add more samples
]

dataset = Dataset.from_dict(samples)

# Run evaluation – this will call retrieve() and generate() internally
results = evaluate(
    dataset,
    retriever=retrieve,
    generator=generate,
    metrics=["retrieval_precision", "retrieval_recall",
             "answer_faithfulness", "answer_relevance",
             "answer_correctness"],
)
print(results.to_dataframe())

The evaluate call is blocking and can be parallelised with Ray or Dask if you have a large test set. In production you’ll likely run it on a scheduled CI job rather than on every request.

How do I evaluate a sample RAG pipeline step-by-step with RAGAS?

Let’s walk through a realistic scenario: a FastAPI endpoint that serves a knowledge-base Q&A. Our pipeline:

Chunk the knowledge base – see my post on Practical RAG chunking strategies for reliable RAG pipelines.
Store embeddings in Pinecone (or any vector DB).
Retrieve top-k chunks per query.
Prompt an LLM with the retrieved context.

We’ll evaluate step 3 (retrieval) and step 4 (generation) together.

# Assume we already have a Pinecone index called `kb-index`
import pinecone
from sentence_transformers import SentenceTransformer

pinecone.init(api_key="YOUR_KEY", environment="us-west1-gcp")
index = pinecone.Index("kb-index")
embedder = SentenceTransformer("all-MiniLM-L6-v2")

def pinecone_retrieve(query: str, k: int = 5) -> List[str]:
    q_vec = embedder.encode([query])[0].tolist()
    hits = index.query(vector=q_vec, top_k=k, include_metadata=True)
    return [hit.metadata["text"] for hit in hits.matches]

Now we plug pinecone_retrieve into the earlier evaluate call, swapping out the simple retrieve function. The rest of the code stays the same.

Running the evaluation on a 200-question test set gave me these scores:

Metric	Score
retrieval_precision	0.78
retrieval_recall	0.71
answer_faithfulness	0.65
answer_relevance	0.73
answer_correctness	0.68
combined	0.71

Interpretation:

Precision is decent, but recall lags. That means we’re missing some crucial evidence. Adding more chunks per query (increase k) or improving the embedding model could help.
Faithfulness is the weakest link. The generator is pulling facts that aren’t present in the retrieved context – classic hallucination. Switching to a more instruction-tuned model or adding a post-generation verification step improves this.
Redundancy (not shown) was 0.42, indicating many overlapping passages. Reducing chunk overlap in the preprocessing stage (see the chunking article) will free up token budget and may lift relevance scores.

How do I interpret the scores and turn them into actionable improvements?

RAGAS scores are percentages, but the absolute number only matters relative to your baseline and SLA. Here’s a quick decision matrix:

Situation	Score range	Action
Retrieval precision < 0.6	Low	Re-train the embedding model or add BM25 fallback.
Retrieval recall < 0.6	Low	Increase `k`, broaden the index (add synonyms), or improve chunk granularity.
Faithfulness < 0.6	Low	Add a "ground-truth check" step: run a second LLM to verify that each sentence is entailed by the context.
Redundancy > 0.5	High	Deduplicate chunks during indexing; use a sliding-window chunker with 30-40% overlap instead of 70%.
Combined score stagnates after tweaks	Stuck	Look at per-question variance – a few outliers can drag the average down. Target those manually.

In my own stack, the biggest win came from context length tuning. I reduced chunk overlap from 70 % to 35 % after the redundancy metric flagged waste. That cut token usage by 22 % and lifted relevance from 0.73 to 0.80 without hurting recall.

Can I automate RAGAS checks in my CI/CD pipeline?

Absolutely. Treat RAGAS like a lint test: fail the build if the combined score drops below a threshold you define (e.g., 0.70). In GitHub Actions, the job looks like this:

name: RAG Pipeline Evaluation

on:
  push:
    branches: [main]
  pull_request:
    branches: [main]

jobs:
  ragas-test:
    runs-on: ubuntu-latest
    steps:
      - uses: actions/checkout@v3
      - name: Set up Python
        uses: actions/setup-python@v4
        with:
          python-version: "3.10"
      - name: Install dependencies
        run: |
          python -m pip install --upgrade pip
          pip install -r requirements.txt
          pip install ragas[all]
      - name: Run RAGAS evaluation
        env:
          OPENAI_API_KEY: ${{ secrets.OPENAI_API_KEY }}
          PINECONE_API_KEY: ${{ secrets.PINECONE_API_KEY }}
        run: |
          python -m ragas.evaluate \
            --dataset tests/rag_test_set.json \
            --threshold 0.70 \
            --output results.json
      - name: Fail if score too low
        run: |
          SCORE=$(jq '.combined' results.json)
          if (( $(echo "$SCORE < 0.70" | bc -l) )); then
            echo "Combined RAGAS score $SCORE below threshold"
            exit 1
          fi

The ragas.evaluate CLI is a thin wrapper around the evaluate function; it prints a JSON report that you can parse. By wiring this into the same pipeline that builds your Docker image for Google Cloud Run (see my Automating Production: A CI/CD Pipeline for Google Cloud Run with GitHub Actions), you guarantee that any regression – whether from a new model version or a changed chunk size – is caught before it reaches users.

Trade-offs and when NOT to use RAGAS

Heavy LLMs in the loop – RAGAS calls your generator for every test case. If you’re evaluating thousands of queries with a costly model, the run can become expensive. In that case, sample a subset or use a cheaper “shadow” model for the evaluation pass.
No gold answers – RAGAS needs ground-truth answers for correctness and faithfulness. For purely exploratory RAG (e.g., internal knowledge bases without labeled data), you can only rely on unsupervised metrics like redundancy, which gives a limited view.
Real-time latency constraints – Running RAGAS in a pre-deployment CI job is fine, but embedding it into an online health-check endpoint adds latency. Keep it offline.

FAQ

Q: Do I need a GPU to run RAGAS?

A: No. Retrieval and metric calculations run on CPU. Only the generator model may need a GPU if you use a large local LLM; otherwise the OpenAI API handles the compute.

Q: Can I customize the weighting of the combined score?

A: Yes. Pass a weights dict to evaluate, e.g., weights={"answer_faithfulness": 0.4, "answer_correctness": 0.4, "retrieval_precision": 0.2}.

Q: How many test examples are enough?

A: For a stable estimate, aim for at least 100 diverse queries covering all intent categories. Smaller sets can be noisy, especially for recall.

Q: Does RAGAS work with multi-turn conversations?

A: It’s primarily designed for single-turn QA. For multi-turn you’d need to flatten the dialogue into a single query or extend the metric suite yourself.

Key Takeaways

rag evaluation using ragas gives you a ready-made, model-agnostic metric suite for RAG pipelines.
Install via pip install ragas[all] and plug in your own retriever and generator functions.
Use the scores to spot concrete issues: low recall → more chunks; low faithfulness → verification step.
Automate the evaluation in CI/CD; fail builds if the combined score falls under your SLA.
Be mindful of cost (LLM calls) and data requirements (gold answers) when scaling the tests.

By treating RAG evaluation as a first-class part of your development workflow, you avoid the hidden debt that comes from “just eyeballing” answers. The next time you add a new embedding model or swap out the LLM, let RAGAS tell you whether you actually improved the user experience.

Comparing AI Agents Python Library Options for Production

Ayush Kumar — Wed, 22 Jul 2026 07:51:03 +0000

Quick answer

If you need a Python library to build an AI agent that can run in production, start with LangChain for flexibility, CrewAI for team-style orchestration, or LlamaIndex if your focus is on data-centric retrieval. All three install with a single pip command, but they differ in abstraction level, runtime cost, and how easy they are to test inside a FastAPI service.

What are the top Python libraries for building AI agents?

The three most battle-tested options today are LangChain, CrewAI, and LlamaIndex.

LangChain gives you granular control over prompts, memory, and tool integration. It’s a good fit when you want to stitch together many LLM calls or custom utilities.
CrewAI treats an agent as a small “crew” of specialists. It’s built around task decomposition, role assignment, and result aggregation. Think of it as a higher-level wrapper that reduces boilerplate for multi-step workflows.
LlamaIndex (formerly GPT Index) shines when your agent needs to search or summarize large document collections. It builds indexes on top of vector stores and lets you query them with natural language.

All three are open source, support OpenAI, Anthropic, and Azure endpoints, and have async APIs that play nicely with FastAPI.

How do I install and set up LangChain vs CrewAI vs LlamaIndex?

Installation is straightforward, but each library pulls a different set of optional dependencies.

# Core libraries
pip install langchain crewai llama-index
# Optional LLM providers (pick what you need)
pip install openai anthropic
# Vector store for LlamaIndex example
pip install chromadb

LangChain minimal setup

from langchain.llms import OpenAI
from langchain.prompts import PromptTemplate
from langchain.chains import LLMChain

llm = OpenAI(model="gpt-4o-mini")
prompt = PromptTemplate.from_template("Translate this to French: {text}")
chain = LLMChain(llm=llm, prompt=prompt)

CrewAI minimal setup

from crewai import Agent, Task, Crew

# Define a simple research agent
researcher = Agent(
    role="Researcher",
    goal="Find the latest Python web frameworks",
    backstory="You are a senior backend engineer.",
    llm="gpt-4o-mini",
)

task = Task(
    description="List three frameworks with a one‑sentence summary each.",
    agent=researcher,
)

crew = Crew(agents=[researcher], tasks=[task])

LlamaIndex minimal setup

from llama_index import VectorStoreIndex, SimpleDirectoryReader
from llama_index.llms import OpenAI

documents = SimpleDirectoryReader("data/").load_data()
index = VectorStoreIndex.from_documents(documents, llm=OpenAI(model="gpt-4o-mini"))

All three snippets run in a fresh virtual environment. The biggest gotcha I’ve hit is version mismatches between langchain and crewai when they both depend on openai. Pinning openai>=1.0,<2.0 in requirements.txt saved me a day of debugging.

Can you show a step-by-step example of a simple AI agent with each library?

Below is a tiny “weather lookup” agent that takes a city name, calls a mock weather API, and returns a friendly sentence. The logic is the same; only the orchestration differs.

LangChain version

import httpx
from langchain.llms import OpenAI
from langchain.prompts import PromptTemplate
from langchain.chains import LLMChain

llm = OpenAI(model="gpt-4o-mini")
prompt = PromptTemplate.from_template(
    "You are a helpful assistant. Use the JSON response {weather} to answer: "
    "What is the weather like in {city}?"
)

async def fetch_weather(city: str) -> dict:
    async with httpx.AsyncClient() as client:
        resp = await client.get(f"https://api.example.com/weather?city={city}")
        return resp.json()

async def weather_agent(city: str) -> str:
    weather = await fetch_weather(city)
    chain = LLMChain(llm=llm, prompt=prompt)
    return await chain.arun(weather=weather, city=city)

CrewAI version

from crewai import Agent, Task, Crew
import httpx

weather_agent = Agent(
    role="Weather Bot",
    goal="Provide a short weather summary",
    backstory="You are a backend engineer who loves concise messages.",
    llm="gpt-4o-mini",
)

async def fetch_weather(city):
    async with httpx.AsyncClient() as client:
        r = await client.get(f"https://api.example.com/weather?city={city}")
        return r.json()

task = Task(
    description="Summarize the JSON weather data for {city}",
    expected_output="A one‑sentence weather report.",
    agent=weather_agent,
    async_func=fetch_weather,   # CrewAI lets you attach a callable
)

crew = Crew(agents=[weather_agent], tasks=[task])

async def run_crew(city):
    return await crew.kickoff(inputs={"city": city})

LlamaIndex version

from llama_index import SimpleDirectoryReader, VectorStoreIndex, ServiceContext
from llama_index.llms import OpenAI
import httpx

# Build a trivial index containing a prompt template
documents = SimpleDirectoryReader(input_files=["prompt.txt"]).load_data()
service_context = ServiceContext.from_defaults(llm=OpenAI(model="gpt-4o-mini"))
index = VectorStoreIndex.from_documents(documents, service_context=service_context)

async def fetch_weather(city):
    async with httpx.AsyncClient() as client:
        r = await client.get(f"https://api.example.com/weather?city={city}")
        return r.json()

async def llama_agent(city):
    query = f"Summarize this JSON: {await fetch_weather(city)}"
    response = await index.as_query_engine().aquery(query)
    return response.response

All three agents return a string like “In Paris it’s 12 °C with light rain.” The LangChain version feels most explicit; CrewAI hides the prompt plumbing but requires you to think in terms of tasks; LlamaIndex bundles the prompt inside an index, which can be overkill for a single API call.

How do the libraries compare on performance and features?

Feature	LangChain	CrewAI	LlamaIndex
Prompt control	Full, low-level	Medium (templates per task)	Low (index-driven)
Tool integration	Easy via `Tool` class	Built-in task-to-tool mapping	Requires custom retriever
Memory	Built-in `ConversationBufferMemory`	Implicit via task history	Index acts as persistent memory
Async support	Native async methods	Async tasks supported	Async query engine
Cost predictability	You manage each LLM call	Crew may fire extra calls for coordination	Index creation can add one-off token cost
Learning curve	Steeper, many concepts	Gentle, opinionated	Gentle if you only need retrieval

In my production services, LangChain adds ~15 ms per extra tool call, CrewAI adds ~30 ms overhead for task orchestration, and LlamaIndex adds a one-time indexing cost that can be several seconds for a 10 k document corpus. If you’re budget-conscious, watch out for CrewAI’s hidden “role-play” calls – they can double your token bill if you forget to set max_tokens on the underlying LLM.

How do I integrate the agent into a FastAPI service?

FastAPI is already async-first, so you can drop any of the async agents directly into a route. Below is a minimal FastAPI app that uses the LangChain weather agent; swap the import for crew_agent or llama_agent to try the other libraries.

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

# Choose your agent implementation
from agents.langchain_weather import weather_agent  # or crew_agent, llama_agent

app = FastAPI(title="Weather AI Agent")

class CityRequest(BaseModel):
    city: str

@app.post("/weather")
async def get_weather(req: CityRequest):
    try:
        result = await weather_agent(req.city)
        return {"city": req.city, "report": result}
    except Exception as exc:
        raise HTTPException(status_code=502, detail=str(exc))

if __name__ == "__main__":
    uvicorn.run("app.main:app", host="0.0.0.0", port=8000, reload=True)

Production tips

Timeouts – LLM calls can stall. Configure httpx timeouts (client = httpx.AsyncClient(timeout=10.0)).
Rate limits – Put a simple token bucket middleware around the route; otherwise you’ll hit OpenAI’s per-minute caps.
Observability – Log the raw LLM request/response (scrub PII) and emit Prometheus metrics for latency and token usage.
Cold start – Load the LLM client once at startup, not per request. I once saw a 2-second spike when I instantiated OpenAI() inside the route handler.

If you need more context, see my earlier post on AI agent python ollama: Build, Test, Deploy with FastAPI for a full deployment pipeline.

What testing and validation strategies work for production agents?

Testing LLM-driven code feels weird because the output is nondeterministic. Here’s what I rely on:

Unit-level prompt snapshots – Store the exact prompt string and expected token count. Use assert "Translate this to French" in prompt. If the prompt changes, the test fails before you even call the model.
Mocked LLM client – Replace OpenAI with a stub that returns a deterministic JSON payload. Libraries like pytest-mock let you patch OpenAI.__call__. This isolates the business logic from the provider.
Contract tests on the FastAPI endpoint – Use httpx.AsyncClient(app=app) in pytest to hit /weather with a fake city and assert the shape of the JSON response.
Canary deployment – Roll out the new agent to 5 % of traffic, compare latency and error rates against the stable version. If the canary’s error rate exceeds 0.5 %, roll back automatically.
Cost guardrails – Write a small background job that queries the OpenAI usage API every hour and alerts when the daily quota is near the limit. I once had a runaway loop that generated 2 M tokens in an hour because a missing max_tokens parameter let the model keep streaming.

When you’re ready to scale, read the “Testing and validation” chapter of the official LangChain docs – they outline a “LLMMock” helper that integrates nicely with pytest.

FAQ

Which library should I start with for a simple chatbot?

LangChain gives the most flexibility and the smallest dependency footprint for a single-turn bot. If you want built-in task management, try CrewAI.

Do I need a vector store for every agent?

Only if you plan to retrieve or embed large text collections. LlamaIndex shines there; otherwise it adds unnecessary latency.

Can I swap OpenAI for an on-prem model?

All three libraries expose a generic LLM interface. Point the client to your local server (e.g., OpenAI(api_base="http://localhost:8000/v1")) and the rest of the code stays the same.

How do I monitor token usage per request?

Wrap the LLM call in a decorator that reads the usage field from the response and pushes it to Prometheus or CloudWatch. I keep a request_id in the FastAPI state to correlate logs.

Key Takeaways

LangChain, CrewAI, and LlamaIndex are the three most production-ready ai agents python library options.
Choose LangChain for fine-grained control, CrewAI for multi-step orchestration, and LlamaIndex for retrieval-heavy workloads.
Installation is a single pip install line; watch version pins for openai.
Each library can be wrapped in an async FastAPI endpoint with minimal boilerplate.
Test prompts, mock LLM calls, and use canary releases to keep production stable.
Keep an eye on latency, token cost, and rate limits – they’re the hidden failure points that bite most engineers.

Event Driven Architecture vs Microservices: When to Use Which

Ayush Kumar — Mon, 20 Jul 2026 10:49:09 +0000

Short answer: Event-driven architecture focuses on asynchronous message flow, while microservices split a system into independently deployable services that usually talk over HTTP. Both solve scalability and organisational challenges, but they do it in different ways and with different failure modes.

In the next few hundred lines I’ll define each style, contrast them, show real-world patterns, and give you Python snippets you can drop into a FastAPI project today. I’ll also flag the hidden costs that bite me most often, so you can decide which approach belongs in your next production stack.

What is event-driven architecture?

Event-driven architecture (EDA) is a design where components react to events - immutable facts that something happened. An event is published once, stored (often in a broker), and any number of consumers may pick it up, process it, and optionally emit new events. The flow is asynchronous by default; the producer never waits for a consumer to finish.

Key traits:

Loose coupling – producers don’t need to know who consumes.
Scalability – add consumers without touching the publisher.
Replayability – stored events can be re-processed for debugging or back-fill.
Eventual consistency – state converges over time, not instantly.

In practice I use Kafka for high-throughput pipelines and Redis Streams for low-latency internal chatter. The broker becomes the system’s nervous system; everything else is just logic that reacts to its pulse.

What is microservices architecture?

Microservices architecture breaks a monolith into a suite of small, independently deployable services. Each service owns its data, its API contract, and its runtime environment. Communication is usually synchronous (HTTP/REST, gRPC) but can also be asynchronous via a message bus.

Core ideas:

Bounded contexts – each service models a specific business capability.
Independent deployment – you can roll out a change without touching the rest.
Technology heterogeneity – different services can use different languages or databases.
Failure isolation – a crash in one service shouldn’t bring the whole system down.

I’ve run dozens of FastAPI-based microservices in production; the pattern works well when the domain can be cleanly divided and you have the ops budget to run multiple containers, health checks, and service meshes.

How do event-driven architecture and microservices differ?

The biggest difference is communication style. In EDA, the contract is an immutable event schema; in microservices, the contract is an API request/response model. That leads to distinct trade-offs:

Aspect	Event-driven	Microservices
Coupling	Very loose; add a consumer without touching the publisher.	Tighter; every API call creates a runtime dependency.
Latency	Usually higher because of broker round-trip and async processing.	Lower for simple request/response paths.
Observability	Need to trace events across topics; replay helps debugging.	Trace HTTP calls; simpler but can be noisy with many services.
Data consistency	Eventual; you must design compensating actions.	Can be strong if each request updates a single service’s DB.
Operational cost	Broker cluster, storage of events, retention policies.	Multiple containers, service discovery, load balancers.
Team autonomy	High; teams can own event producers or consumers independently.	High; each team owns a service end-to-end.

When you mix the two (microservices that publish/consume events) you get the best of both worlds, but you also inherit the combined complexity.

When should I pick event-driven over microservices (or vice versa)?

Pick event-driven architecture when:

You need to fan-out work to many independent workers (e.g., image processing, model inference).
Data arrives as a stream and you must guarantee no loss (financial tick data, IoT telemetry).
Business processes span multiple domains and you want a single source of truth for “what happened”.

Pick microservices when:

Your domain can be cleanly split into bounded contexts with well-defined APIs.
Low latency is a hard requirement (e.g., real-time pricing engine).
You have the ops bandwidth to manage many services, health checks, and versioned APIs.

Don’t force a hybrid if you only have a handful of services and a simple queue would do. Adding a broker just to “future-proof” the design often adds unnecessary operational debt.

Real-world examples and integration patterns

1. Order processing pipeline

Event-driven: An order_created event lands on Kafka. A payment service consumes it, attempts a charge, and publishes payment_succeeded or payment_failed. A shipping service listens for payment_succeeded and creates a shipment. If any step fails, a compensating order_canceled event rolls back the process.
Microservices: The frontend calls the order service via HTTP, which internally calls the payment service synchronously, then the shipping service. If payment fails, the order service returns an error immediately.

The event-driven version tolerates spikes in order volume because each consumer can autoscale independently. The microservice version is simpler to reason about for a low-traffic shop.

2. Feature flag rollout

Publish a feature_flag_updated event whenever a flag changes in the admin UI.
All services subscribe; they update their in-memory cache without restarting.
No need for a central config server hitting every service on each request.

3. AI inference as a service

A FastAPI endpoint receives a request, writes the payload to a inference_requested topic, and returns a 202 Accepted.
Workers pull the event, run the model, and publish inference_completed.
The original client polls a /status/{id} endpoint or receives a webhook.

This pattern decouples the heavy GPU work from the HTTP layer and lets you scale workers separately. See my post on building a robust RAG pipeline for more details.

Best practices for Python implementations

Below is a minimal, production-ready skeleton that shows both styles side-by-side. I use FastAPI, Pydantic, and Redis Streams for the event bus; you can swap Kafka with a few lines.

# app/main.py
from fastapi import FastAPI, BackgroundTasks, HTTPException
from pydantic import BaseModel, Field
import redis
import json
import uuid

app = FastAPI()
r = redis.Redis(host="redis", port=6379, decode_responses=True)

STREAM = "orders"

class OrderIn(BaseModel):
    user_id: int = Field(..., gt=0)
    amount: float = Field(..., gt=0)

# ------------------- Event‑driven endpoint -------------------
@app.post("/orders/async")
async def create_order_async(order: OrderIn, bg: BackgroundTasks):
    """
    Publishes an order_created event and returns immediately.
    """
    event = {
        "id": str(uuid.uuid4()),
        "type": "order_created",
        "payload": order.dict(),
    }
    # Write to Redis Stream; * means auto‑generated ID
    r.xadd(STREAM, event)
    return {"order_id": event["id"], "status": "queued"}

# ------------------- Synchronous microservice endpoint -------------------
@app.post("/orders/sync")
async def create_order_sync(order: OrderIn):
    """
    Calls the payment microservice synchronously and returns the result.
    """
    # Simulate an HTTP call to payment service
    payment_ok = await call_payment_service(order)
    if not payment_ok:
        raise HTTPException(status_code=402, detail="Payment failed")
    # Persist order in DB (omitted)
    return {"order_id": str(uuid.uuid4()), "status": "confirmed"}

async def call_payment_service(order: OrderIn) -> bool:
    # In real code use httpx or aiohttp; here we just fake it
    return order.amount < 1000  # simple rule for demo

Consumer worker (event-driven)

# workers/payment_worker.py
import redis, json, time

r = redis.Redis(host="redis", port=6379, decode_responses=True)
STREAM = "orders"

def process_payment(event):
    payload = json.loads(event["payload"])
    # Insert real payment logic here
    success = payload["amount"] < 5000
    result = {
        "type": "payment_succeeded" if success else "payment_failed",
        "order_id": event["id"],
        "payload": payload,
    }
    r.xadd("payments", result)

while True:
    msgs = r.xread({STREAM: "0"}, block=5000, count=10)
    if not msgs:
        continue
    for _, entries in msgs:
        for entry_id, entry in entries:
            process_payment(entry)
    time.sleep(0.1)

Key takeaways from the code

Keep event payloads small and serializable (JSON works, but Avro/Protobuf help with schema evolution).
Use a background task or separate worker process; never block the FastAPI request thread.
Store the event ID in your DB as the source of truth; it lets you replay safely.
For microservice calls, wrap HTTP clients with retries and circuit-breakers (see the httpx docs).

Observability tips

Tracing – instrument both producers and consumers with OpenTelemetry. Tag events with correlation_id so you can stitch a timeline across services.
Dead-letter queues – configure a separate stream for events that repeatedly fail. Don’t let a poison pill stall the whole pipeline.
Schema registry – if you move to Kafka, a Confluent-style registry prevents breaking changes.

When NOT to use this pattern

If your team can’t maintain a broker, the whole system becomes a single point of failure.
When you need sub-millisecond latency (high-frequency trading), the extra hop kills you.
If you only have one consumer, a simple task queue (Celery) may be enough.

FAQ

Q: Can I run both patterns in the same codebase?

A: Absolutely. Many teams expose HTTP endpoints for CRUD while publishing domain events for downstream processing. Just keep the responsibilities clear: APIs for request/response, events for side-effects.

Q: Do I need a separate database for each microservice?

A: Not strictly, but sharing a DB creates hidden coupling. If you store data in a single relational DB, treat it as a shared resource and accept the risk of cross-service transactions.

Q: How do I version event schemas?

A: Use a schema registry or embed a version field in the event payload. Consumers should be tolerant of unknown fields and reject events with unsupported versions.

Q: What monitoring metrics matter most?

A: For brokers: lag, consumer throughput, and error rate. For services: request latency, error ratio, and circuit-breaker state.

Key Takeaways

Event-driven architecture = async, loosely coupled, eventual consistency; microservices = sync API contracts, independent deployment, often strong consistency.
Choose EDA for fan-out workloads, high-volume streams, or when you need replayability. Choose microservices for low-latency, clear bounded contexts, and when you can afford multiple containers.
Mixing both is common: microservices publish events, other services consume them.
In Python, FastAPI + Redis Streams or Kafka gives a clean separation; keep workers out of the request thread and use OpenTelemetry for tracing.
Watch operational costs: broker clusters, schema management, and dead-letter handling can become the hidden budget line items.

Happy building, and feel free to drop a comment if you run into a specific edge case.

AI agent python ollama: Build, Test, Deploy with FastAPI

Ayush Kumar — Fri, 17 Jul 2026 09:08:29 +0000

I’ll show you how to get an ai agent python ollama up and running on your laptop, expose it through FastAPI, test it automatically, and push it to production. You can have a working endpoint in under an hour if you follow the steps carefully.

How do I install and configure Ollama for local LLM inference?

The first thing you need is a running Ollama server. It ships as a single binary for Linux, macOS, and Windows, so there are no heavyweight dependencies.

# Download the latest release (replace with the version you need)
curl -L https://ollama.com/download/ollama-linux-amd64 -o ollama
chmod +x ollama
sudo mv ollama /usr/local/bin/

Start the daemon:

ollama serve &

Ollama defaults to listening on http://127.0.0.1:11434. You can change the bind address with the --host flag or set a custom port with --port. I keep it on the default because my FastAPI container can reach the host network via Docker’s host mode.

Next, pull a model. Ollama supports many open-source LLMs; for a quick start I use phi3 because it fits in 2 GB of RAM.

ollama pull phi3

If you’re on a low-memory box, add --cpu to force CPU inference. It’s slower, but it avoids OOM kills that have bitten me on cheap VPS instances.

Configuration tip: create a ~/.ollama/config.yaml:

host: "0.0.0.0"
port: 11434
log_level: "info"

That way you can restart the service without remembering flags.

How can I build an AI agent in Python using Ollama’s REST API?

Ollama exposes a simple JSON-over-HTTP API. The core endpoint is POST /api/chat. Here’s a minimal wrapper that turns the HTTP call into a Python class.

import httpx
from typing import List, Dict

class OllamaChat:
    def __init__(self, model: str = "phi3", base_url: str = "http://127.0.0.1:11434"):
        self.base_url = base_url.rstrip("/")
        self.model = model
        self.client = httpx.Client(timeout=30.0)

    def chat(self, messages: List[Dict[str, str]]) -> str:
        payload = {
            "model": self.model,
            "messages": messages,
            "stream": False
        }
        resp = self.client.post(f"{self.base_url}/api/chat", json=payload)
        resp.raise_for_status()
        return resp.json()["message"]["content"]

The messages list follows the OpenAI chat format (role = system|user|assistant). Because Ollama mimics that schema, you can reuse prompt engineering tricks you already know.

Agent pattern: I like to encapsulate a single “goal” function. For example, a simple calculator agent:

def calculator_agent(question: str) -> str:
    wrapper = OllamaChat()
    msgs = [
        {"role": "system", "content": "You are a reliable calculator. Return only the numeric answer."},
        {"role": "user", "content": question}
    ]
    return wrapper.chat(msgs)

When I first built this, I forgot to set stream: False and got a generator back, which broke downstream code that expected a string. Always verify the response shape in a unit test.

How do I integrate the Ollama-powered agent into a FastAPI service?

FastAPI makes wiring an endpoint trivial. Below is a complete main.py that exposes /calculate and forwards the request to the calculator_agent.

from fastapi import FastAPI, HTTPException
from pydantic import BaseModel
from typing import Literal

app = FastAPI(title="Ollama AI Agent Service")

class CalcRequest(BaseModel):
    expression: str

class CalcResponse(BaseModel):
    result: str

@app.post("/calculate", response_model=CalcResponse)
def calculate(req: CalcRequest):
    try:
        answer = calculator_agent(req.expression)
    except httpx.HTTPError as exc:
        raise HTTPException(status_code=502, detail=f"Ollama error: {exc}") from exc
    return CalcResponse(result=answer)

Run it with:

uvicorn main:app --host 0.0.0.0 --port 8000

If you’re containerizing, keep the Ollama server in a sidecar container and share a Docker network. I once tried to run Ollama inside the same container as FastAPI, but the process died when the container restarted, leaving the API hanging. Separate containers give you independent restart policies.

Production tip: enable uvicorn workers (--workers 4) to utilize multiple CPU cores. Ollama itself is single-threaded per model, so you’ll want a pool of models or a load balancer if you expect high QPS.

How do I automate validation and testing of the agent’s actions?

Testing LLM-driven code is tricky because the output is nondeterministic. I approach it in two layers:

Prompt contract tests – assert that the response adheres to a JSON schema or a regex pattern.
Functional integration tests – spin up a real Ollama instance (via Docker) and run end-to-end requests.

Here’s a pytest fixture that starts Ollama in a temporary container:

import subprocess
import time
import pytest

@pytest.fixture(scope="session")
def ollama_server():
    proc = subprocess.Popen(["docker", "run", "--rm", "-p", "11434:11434", "ollama/ollama"], stdout=subprocess.PIPE)
    # Wait for the health endpoint
    for _ in range(10):
        try:
            httpx.get("http://127.0.0.1:11434/health").raise_for_status()
            break
        except Exception:
            time.sleep(1)
    else:
        pytest.fail("Ollama did not start in time")
    yield
    proc.terminate()

And a test that validates the calculator:

def test_calculator_agent(ollama_server):
    result = calculator_agent("What is 7 * 8?")
    assert result.strip() == "56"

If the model ever drifts (e.g., it starts adding explanations), the test will fail, alerting you before you ship.

When NOT to rely on this: for creative generation (storytelling, code synthesis) you can’t lock down exact output. In those cases, check for presence of required fields rather than exact string equality.

How do I deploy and scale the Ollama agent in production?

Running Ollama on a single machine works for prototypes, but production traffic often exceeds a single CPU’s capacity. My current stack uses three layers:

Model shards – run multiple Ollama containers, each loading the same model but pinned to a different CPU core using taskset.
Reverse proxy – an Nginx or Traefik instance routes /api/chat requests round-robin to the shards.
FastAPI autoscaling – Kubernetes Horizontal Pod Autoscaler (HPA) watches the FastAPI service’s request latency and spins up more pods as needed.

A minimal Kubernetes deployment for Ollama looks like:

apiVersion: apps/v1
kind: Deployment
metadata:
  name: ollama
spec:
  replicas: 3
  selector:
    matchLabels:
      app: ollama
  template:
    metadata:
      labels:
        app: ollama
    spec:
      containers:
        - name: ollama
          image: ollama/ollama:latest
          ports:
            - containerPort: 11434
          resources:
            limits:
              cpu: "2"
              memory: "4Gi"
          command: ["ollama", "serve", "--host", "0.0.0.0", "--port", "11434"]

Combine that with a Service:

apiVersion: v1
kind: Service
metadata:
  name: ollama
spec:
  selector:
    app: ollama
  ports:
    - protocol: TCP
      port: 11434
      targetPort: 11434

Your FastAPI pod just points to http://ollama:11434. If a node runs out of memory, the pod restarts – I’ve added a livenessProbe that hits /health to catch silent OOM kills.

Cost considerations: each model instance reserves RAM for the weights. A 3-B parameter model needs ~6 GB. Running three shards can easily hit 20 GB, so budget accordingly. If you’re on a spot-instance fleet, be prepared for occasional restarts; make your FastAPI client retry with exponential back-off.

When not to scale this way: if you need sub-millisecond latency, Ollama’s CPU inference isn’t enough. In that case, move to a GPU-backed service or a hosted provider that offers hardware acceleration.

FAQ

What language models does Ollama support?

Ollama ships with many open-source models: Llama 3, Mistral, Phi-3, and community-built variants. You pull them with ollama pull <model> and they run on CPU or GPU depending on your hardware.

Can I use the same code with OpenAI’s API?

Yes. The request schema (model, messages, stream) mirrors OpenAI’s chat endpoint, so swapping the base URL and API key converts the wrapper to an OpenAI client. See the related post “Using the OpenAI Agents SDK for Python: A Practical Guide”.

Do I need to restart Ollama after changing the model?

No. After pulling a new model you can start using it immediately by passing the model name to OllamaChat. The server loads the weights lazily on the first request.

How do I monitor Ollama’s performance?

Expose the /metrics endpoint (Prometheus format) by running Ollama with --metrics. Then scrape it with Grafana or Prometheus to watch request latency, CPU usage, and cache hits.

Key Takeaways

Install Ollama once, pull a lightweight model, and you have a local LLM server ready for HTTP calls.
Wrap the /api/chat endpoint in a thin Python class; it gives you a reusable ai agent python ollama building block.
FastAPI integration is a few lines of code; keep Ollama in a sidecar or separate pod for resilience.
Automated tests that spin up a temporary Ollama container catch prompt drift before it reaches users.
Scale by sharding model instances behind a reverse proxy and let Kubernetes handle FastAPI pod autoscaling.
Watch memory footprints; each model instance eats several gigabytes, so plan resources or consider GPU alternatives for low-latency workloads.

With these pieces in place, you can move from a local experiment to a production-grade AI agent service without reinventing the wheel. Happy coding!

Building microservices python fastapi: design to production

Ayush Kumar — Tue, 14 Jul 2026 14:08:36 +0000

Why microservices python fastapi are a solid choice for production

If you need to spin up a set of loosely-coupled services that can evolve independently, microservices python fastapi is a pragmatic answer. FastAPI gives you async-first request handling, automatic OpenAPI docs, and type-safe validation with almost no boilerplate. Coupled with Python’s rich ecosystem for data, AI, and messaging, you can ship production-grade services in weeks instead of months.

In this post I’ll walk through the entire lifecycle: designing a FastAPI-based microservice, picking a communication pattern, wiring RabbitMQ or Kafka, containerizing with Docker Compose and Kubernetes, testing, monitoring, tracing, and locking down security. I’ll also sprinkle in the hard-earned lessons that kept me awake at 2 am.

How do I design a FastAPI-based microservice that stays maintainable?

The first step is to define a clear bounded context. Each service should own a single business capability and expose a minimal public API. I start with a folder layout that mirrors that intent:

myservice/
├── app/
│   ├── __init__.py
│   ├── api/
│   │   ├── v1/
│   │   │   └── endpoints.py
│   ├── core/
│   │   ├── config.py
│   │   └── di.py          # dependency injection helpers
│   ├── models/
│   │   └── orm.py
│   └── services/
│       └── business.py
├── tests/
│   └── test_endpoints.py
├── Dockerfile
└── pyproject.toml

api/v1/endpoints.py holds the router objects, nothing else.
services/business.py contains pure Python functions that implement the domain logic. No FastAPI imports.
core/di.py wires the dependencies (DB session, external clients) using FastAPI’s Depends.

Why separate business logic from the router? Because it lets you unit-test the core without starting an ASGI server and keeps the codebase friendly to future refactors (e.g., swapping FastAPI for another framework).

A minimal router looks like this:

# app/api/v1/endpoints.py
from fastapi import APIRouter, Depends, HTTPException, status
from ..services.business import greet_user
from ..core.di import get_db

router = APIRouter(prefix="/v1", tags=["greeting"])

@router.get("/hello/{name}", response_model=str)
async def hello(name: str, db=Depends(get_db)):
    try:
        return await greet_user(name, db)
    except ValueError as exc:
        raise HTTPException(status_code=status.HTTP_400_BAD_REQUEST,
                            detail=str(exc))

Notice the thin wrapper around greet_user. The service function can be called from a Celery task, a CLI script, or a test harness without any FastAPI baggage.

Lesson learned: I once let the router import the ORM models directly and ended up with circular imports that broke hot-reloading in development. Keeping the layers strictly separated avoided that mess.

Which communication pattern should I pick: REST, gRPC, or async messaging?

When is REST enough?

If the interaction is request-response and latency tolerance is in the hundreds of milliseconds, plain HTTP/JSON works fine. FastAPI’s automatic OpenAPI spec makes client generation painless.

Pros:

Human-readable payloads, easy debugging with curl.
No extra runtime (just uvicorn).

Cons:

Verbose JSON for high-frequency data.
No built-in streaming or binary support.

When does gRPC make sense?

For internal, high-throughput services that need strict contracts and binary payloads, gRPC shines. FastAPI can still expose a REST side-car while a separate gRPC server handles the heavy lifting.

Pros:

Protobuf enforces schema, reduces payload size.
Built-in code generation for many languages.

Cons:

Requires a separate server process or grpcio integration.
Debugging is less straightforward; you need grpcurl or similar.

When to use async messaging (Kafka, RabbitMQ)?

If you need eventual consistency, event sourcing, or fan-out to many consumers, an async broker is the answer. You can still expose a REST endpoint that publishes a message, but the heavy work happens downstream.

Trade-off: Async adds operational complexity (broker management, duplicate handling) and latency (typically seconds). Don’t use it for simple CRUD that needs immediate confirmation.

My rule of thumb:

Start with REST.
Add gRPC only if you hit protobuf-friendly performance limits.
Introduce a message queue when you need decoupling or replayability.

How do I implement message queues with RabbitMQ or Kafka?

FastAPI itself is agnostic; the integration lives in background workers. I prefer RabbitMQ for task queues (Celery) and Kafka for event streams.

RabbitMQ + Celery example

# app/core/celery_app.py
from celery import Celery

celery = Celery(
    "myservice",
    broker="amqp://guest:guest@rabbitmq:5672//",
    backend="redis://redis:6379/0",
)

@celery.task
def process_order(order_id: int):
    # heavy processing, DB writes, external calls
    ...

# app/api/v1/endpoints.py
from ..core.celery_app import process_order

@router.post("/orders")
async def create_order(order: OrderIn):
    # persist order synchronously
    db_order = await save_order(order)
    # fire‑and‑forget async work
    process_order.delay(db_order.id)
    return {"id": db_order.id}

Failure mode: If the Celery worker crashes after the DB commit but before the task is enqueued, you lose the async step. To mitigate, use the Outbox pattern: write an “outbox” table in the same transaction, and have a separate poller publish those rows to RabbitMQ.

Kafka producer example

# app/services/events.py
from aiokafka import AIOKafkaProducer
import json
import os

producer = AIOKafkaProducer(
    bootstrap_servers=os.getenv("KAFKA_BOOTSTRAP_SERVERS")
)

async def publish_user_created(user_id: int):
    await producer.start()
    try:
        await producer.send_and_wait(
            "user.created",
            json.dumps({"user_id": user_id}).encode("utf-8")
        )
    finally:
        await producer.stop()

Kafka guarantees ordering within a partition and retains messages for replay. The downside is higher operational overhead: you need Zookeeper/KRaft, topic configurations, and careful consumer offset management.

When NOT to use: If you only have a handful of services and can tolerate occasional coupling, adding Kafka is overkill. It also hurts latency for real-time UI updates; a WebSocket or SSE may be better.

How do I containerize and orchestrate a FastAPI microservice with Docker Compose & Kubernetes?

Dockerfile (multi-stage)

# syntax=docker/dockerfile:1
FROM python:3.12-slim AS builder
WORKDIR /app
COPY pyproject.toml poetry.lock ./
RUN pip install poetry && poetry export -f requirements.txt --output requirements.txt --without-hashes
RUN pip install --no-cache-dir -r requirements.txt

FROM python:3.12-slim AS runtime
WORKDIR /app
COPY --from=builder /usr/local/lib/python3.12/site-packages /usr/local/lib/python3.12/site-packages
COPY . .
EXPOSE 8000
CMD ["uvicorn", "app.main:app", "--host", "0.0.0.0", "--port", "8000"]

The builder stage isolates pip install, keeping the final image tiny (≈80 MB).

Docker Compose for local dev

version: "3.9"
services:
  api:
    build: .
    ports:
      - "8000:8000"
    environment:
      - DATABASE_URL=postgresql://postgres:postgres@db:5432/mydb
    depends_on:
      - db
      - rabbitmq
  db:
    image: postgres:16-alpine
    environment:
      POSTGRES_USER: postgres
      POSTGRES_PASSWORD: postgres
      POSTGRES_DB: mydb
    volumes:
      - pgdata:/var/lib/postgresql/data
  rabbitmq:
    image: rabbitmq:3-management
    ports:
      - "5672:5672"
      - "15672:15672"
volumes:
  pgdata:

Compose gives me a one-command dev environment. I’ve been bitten by mismatched environment variables when moving to Kubernetes, so I keep a single source of truth in a .env file and reference it both locally and in the Helm chart.

Helm chart snippet for Kubernetes

apiVersion: apps/v1
kind: Deployment
metadata:
  name: {{ include "myservice.fullname" . }}
spec:
  replicas: {{ .Values.replicaCount }}
  selector:
    matchLabels:
      app.kubernetes.io/name: {{ include "myservice.name" . }}
  template:
    metadata:
      labels:
        app.kubernetes.io/name: {{ include "myservice.name" . }}
    spec:
      containers:
        - name: api
          image: "{{ .Values.image.repository }}:{{ .Values.image.tag }}"
          ports:
            - containerPort: 8000
          envFrom:
            - secretRef:
                name: myservice-secret
          readinessProbe:
            httpGet:
              path: /healthz
              port: 8000
            initialDelaySeconds: 5
            periodSeconds: 10

Kubernetes gives you automated rollouts, pod health checks, and horizontal scaling. The only surprise I ran into was the default termination grace period (30 s) being too short for long-running DB migrations. I increased terminationGracePeriodSeconds to 60 s in the pod spec.

How should I test, monitor, and trace FastAPI microservices?

Unit & integration testing

I keep the business logic free of FastAPI, so a typical unit test looks like:

# tests/test_business.py
import pytest
from app.services.business import greet_user

@pytest.mark.asyncio
async def test_greet_user():
    result = await greet_user("Alice", db=None)  # db mock or None for pure logic
    assert result == "Hello, Alice!"

For endpoint tests I use httpx with FastAPI’s TestClient:

from httpx import AsyncClient
from app.main import app

@pytest.mark.asyncio
async def test_hello_endpoint():
    async with AsyncClient(app=app, base_url="http://test") as client:
        resp = await client.get("/v1/hello/Bob")
    assert resp.status_code == 200
    assert resp.json() == "Hello, Bob!"

I once faced a SQLAlchemy session leak that only manifested after hours of load. The fix is detailed in my FastAPI SQLAlchemy Session Leak Detection post, but the quick tip is: always use async with get_db() as db: in background tasks, and never store the session globally.

Monitoring & tracing

Prometheus + Grafana for metrics. FastAPI exposes /metrics through prometheus_fastapi_instrumentator.
OpenTelemetry for distributed tracing. I instrument the HTTP layer and the Kafka producer:

from opentelemetry.instrumentation.fastapi import FastAPIInstrumentor
from opentelemetry.instrumentation.aiokafka import AIOKafkaInstrumentor

FastAPIInstrumentor().instrument_app(app)
AIOKafkaInstrumentor().instrument()

Health checks: /healthz endpoint returns DB and broker status. Kubernetes uses it for liveness.

A common pitfall: forgetting to set OTEL_EXPORTER_OTLP_ENDPOINT in the container, which made tracing silently drop. Adding the env var to the Helm values fixed it.

Load testing

I run locust against the /v1/hello/{name} endpoint with 200 concurrent users. The latency stayed under 150 ms, well within my SLA. When I increased to 500 users, the CPU spiked to 90 % and the service started returning 502 from the ingress. The fix was to enable Uvicorn workers (uvicorn app.main:app --workers 4) and tune the Gunicorn timeout.

What are the security and authentication best practices for FastAPI microservices?

Prefer OAuth2 with JWT for inter-service auth. FastAPI’s OAuth2PasswordBearer combo works, but for service-to-service I use client credentials flow and validate the token with a shared public key.

from fastapi import Security, HTTPException
from fastapi.security import HTTPBearer, HTTPAuthorizationCredentials
from jose import jwt

bearer = HTTPBearer()

def verify_token(credentials: HTTPAuthorizationCredentials = Security(bearer)):
    try:
        payload = jwt.decode(credentials.credentials, PUBLIC_KEY, algorithms=["RS256"])
        return payload
    except jwt.JWTError:
        raise HTTPException(status_code=401, detail="Invalid token")

Scope checks: embed service name and allowed actions in the token claims, and verify them in each endpoint.
Rate limiting: Deploy Envoy as a sidecar with a token-bucket filter. It protects against accidental DoS from internal clients.
Input validation: FastAPI’s Pydantic models already reject malformed JSON, but never trust external libraries. I once let a third-party library deserialize raw JSON into a dict, opening a prototype pollution path. The fix was to keep all external data behind a Pydantic model.
Secrets management: Store DB passwords, API keys, and JWT signing keys in Kubernetes Secrets or HashiCorp Vault. Never hard-code them. In my CI pipeline I use kubectl create secret generic with --from-literal.
CORS: Only allow origins that belong to your front-end. app.add_middleware(CORSMiddleware, allow_origins=["https://myapp.com"], ...).
Patch dependencies: Run pip list --outdated in CI and automate security scans with GitHub Dependabot. A recent CVE in pyyaml forced me to upgrade across all services within a day.

FAQ

Q: Should I use async SQLAlchemy with FastAPI?

A: Yes, if your service does I/O-bound DB work. Async drivers (asyncpg) avoid thread-pool exhaustion. Just remember to close the session in a finally block or use a dependency that yields the session.

Q: When is it safe to run multiple FastAPI workers in the same pod?

A: When the service is CPU-bound or you need higher throughput. Use a process manager like Gunicorn with the uvicorn.workers.UvicornWorker. Avoid sharing in-memory caches between workers; use Redis instead.

Q: How do I avoid “QueuePool limit reached” errors?

A: Tune the SQLAlchemy pool_size and max_overflow settings, and make sure every request returns the DB connection to the pool. My write-up on fixing that error is Fixing “QueuePool limit reached”.

Q: Do I need OpenAPI docs in production?

A: They’re useful for internal debugging, but expose them only behind authentication or disable them with docs_url=None in production to reduce surface area.

Key Takeaways

Design first: isolate business logic from FastAPI routers to keep tests fast and code reusable.
Pick the right communication pattern: start with REST, add gRPC or async messaging only when the use case demands it.
Use RabbitMQ for tasks, Kafka for event streams; guard against outbox consistency issues.
Containerize with multi-stage Docker, run locally with Docker Compose, and ship to Kubernetes via Helm.
Test early, monitor with Prometheus/OpenTelemetry, and watch for session leaks (see my session-leak guide).
Secure everything: JWT verification, scope checks, rate limiting, secret management, and regular dependency audits.

Microservices python fastapi isn’t a silver bullet, but with the right patterns you can ship reliable, observable, and secure services that survive the chaos of production. Happy coding!

Why FastAPI Apps Slow Down Over Time (Low CPU, High Latency Explained)

Ayush Kumar — Tue, 14 Jul 2026 13:55:19 +0000

The "Silent Death" Scenario

You deploy your FastAPI application. For the first few hours, it feels blazing fast. Your P99 latency is sitting comfortably at 20ms. You high-five your team and go to sleep.

The next morning, your alerts are screaming.

Latency: Spiking to 3,000ms+ randomly.
Health Checks: Failing intermittently.
The Weird Part: You check your dashboards, and CPU usage is at 5%. Memory is flat. The database is responding instantly.

If your server isn't out of CPU and isn't out of memory, why is it crawling?

The answer lies in The Event Loop. When CPU is low but latency is high, it usually means your application isn't working, it is waiting.

This guide focuses on real-world scenarios where FastAPI apps slow down over time despite low CPU usage, a pattern that only appears under sustained load.

TL;DR: If your FastAPI app has low CPU but high latency, your event loop is blocked. The fix is not scaling CPUs, it’s fixing lifecycle, async misuse, and resource ownership.

Understanding the Single-Threaded Model

To fix the problem, you must first understand the machine.

FastAPI (running on Uvicorn) relies on Python's asyncio event loop. Think of the Event Loop as a single waiter in a busy restaurant.

Async: When a customer orders food (an API request), the waiter sends the order to the kitchen (Database/External API) and immediately goes to serve the next table. He doesn't wait in the kitchen. This is why FastAPI can handle thousands of requests with one thread.
Sync (Blocking): If the waiter decides to cook the food himself (Heavy Calculation) or stand in the kitchen staring at the chef (Synchronous I/O), he ignores all other tables.

If that waiter stops for just 1 second to chop an onion, every other customer waits 1 second longer. In a high-throughput system, these delays stack up exponentially.

**Mental Model**  
Think of the event loop as a single-lane highway:
- Async = cars exit quickly, highway stays free
- Blocking = one broken car stops everything behind it

Killer #1: The "Sync" Blocker (and the `def` Trap)

This is the most common mistake developers make when migrating from Flask or Django.

In FastAPI, the moment you define an endpoint with async def, you are promising FastAPI: "I will not block the main thread." FastAPI trusts you and runs that code directly on the main event loop.

If you put anything blocking inside an async def, you freeze the entire server.

The Mistake: Synchronous I/O in Async Path

Python

import time
import requests  # <--- The silent killer

@app.get("/external-data")
async def get_data():
    # BLOCKING!
    # This freezes the entire server. No other request can enter.
    response = requests.get("https://example.com/huge-data")
    time.sleep(1) # Simulating processing
    return response.json()

If 50 users hit this endpoint simultaneously, the 50th user doesn't wait 1 second. They wait 50 seconds (1s x 50). The CPU usage remains low because the CPU isn't computing - it's just idle, waiting for the network response.

The Nuance: `def` vs `async def`

FastAPI has a safety valve. If you define your endpoint with just def (no async), FastAPI assumes it might block and runs it in a separate thread pool (Starlette's run_in_threadpool).

Python

@app.get("/safe-sync")
def safe_sync():
    # Safe! FastAPI runs this in a thread pool.
    time.sleep(1)
    return {"status": "ok"}

However, you lose the benefits of async concurrency. The best solution is to keep async def but offload the blocking work.

💡
Checkout that you are using your wokers wisely, so it doesn’t slow down your requests.

The Fix: `asyncio.to_thread`

If you must use a blocking library (like requests or pandas) or perform a heavy calculation (like image processing), run it in a non-blocking way.

Python

import asyncio
import time

@app.get("/heavy-async")
async def heavy_async():
    # ✅ Runs in a separate thread, leaving the Event Loop free
    await asyncio.to_thread(time.sleep, 1)
    return {"status": "done"}

Red Flags That Your Event Loop Is Blocked

If you see any of these, suspect blocking immediately:

Health checks timing out while business endpoints are slow
Latency spikes without CPU spikes
Restarting pods “fixes” performance temporarily
Throughput collapses before memory or CPU exhaustion

Killer #2: The Hidden CPU Crushers (Password Hashing)

Sometimes, the blocker isn't I/O; it's pure CPU work.

A classic example is Password Hashing (bcrypt or argon2). Hashing a password is designed to be slow and CPU-intensive to thwart hackers.

The Mistake:

Python

from passlib.context import CryptContext

pwd_context = CryptContext(schemes=["bcrypt"], deprecated="auto")

@app.post("/login")
async def login(user: UserLogin):
    # BLOCKING! 
    # Verifying a hash takes ~300ms of pure CPU time.
    if not pwd_context.verify(user.password, user.hashed_password):
        raise HTTPException(...)
    return {"token": "..."}

If 10 users try to login at once, your event loop locks up for 3 seconds (300ms * 10). Your health check endpoint (which should take 1ms) will time out, causing Kubernetes to kill your pod.

Password hashing belongs to the request lifecycle, but not to the event loop.

The Fix: Even though this is CPU-bound, not I/O-bound, you must still move it off the main thread.

Python

@app.post("/login")
async def login(user: UserLogin):
    # Offload CPU work to a thread
    is_valid = await asyncio.to_thread(
        pwd_context.verify, user.password, user.hashed_password
    )
    if not is_valid:
        raise HTTPException(...)

Killer #3: The `httpx` Client Leak

If your API acts as a gateway or aggregator calling other services, you are likely using httpx or aiohttp. A common mistake is treating the async client like a standard variable.

HTTP clients should be treated like database engines: created once, owned by the application lifecycle, and cleaned up deterministically.

The Mistake: Creating a Client per Request

Python

@app.get("/proxy")
async def proxy_request():
    # Creates a new Connection Pool + SSL Context every time!
    async with httpx.AsyncClient() as client:
        resp = await client.get("https://example.com")
    return resp.json()

Why this kills performance:

TCP Handshake: Every request performs a 3-way handshake.
SSL Handshake: Expensive crypto negotiation happens every time.
File Descriptors: You will eventually run out of sockets (OSError: [Errno 24] Too many open files).

The Fix: The Lifespan Pattern Create one client at startup and reuse it. The connection pool stays open, making subsequent requests lightning fast.

Python

from contextlib import asynccontextmanager

@asynccontextmanager
async def lifespan(app: FastAPI):
    # Initialize one client for the whole app
    app.state.http_client = httpx.AsyncClient()
    yield
    # Cleanup on shutdown
    await app.state.http_client.aclose()

app = FastAPI(lifespan=lifespan)

@app.get("/proxy")
async def proxy_request(request: Request):
    client = request.state.http_client
    resp = await client.get("https://example.com")
    return resp.json()

Killer #4: The "Zombie" Background Task

FastAPI provides BackgroundTasks, which are great. But often, developers want "fire and forget" logic using standard AsyncIO.

The Mistake:

Python

@app.post("/process")
async def process():
    # DANGER! 
    # The Python Garbage Collector might destroy this task 
    # immediately because no variable references it.
    asyncio.create_task(long_running_job()) 
    return {"status": "started"}

Not only can this task fail silently, but if it hangs, you have no way to cancel it or monitor it.

The Fix: The Set-Reference Pattern You must maintain a "Hard Reference" to your running tasks.

Python

# A global set to hold references
active_tasks = set()

async def safe_process():
    task = asyncio.create_task(long_running_job())

    # 1. Add to set (keep it alive)
    active_tasks.add(task)

    # 2. Remove from set when done (prevent memory leak)
    task.add_done_callback(active_tasks.discard)

Why Adding More CPU or Pods Won’t Save You

Auto-scaling hides event loop bugs, it doesn’t fix them.

When the loop is blocked:

Each new pod reproduces the same problem
Load balancers distribute latency, not relief
Costs go up while performance stays bad

Until the event loop is unblocked, scaling is cosmetic.

How to Debug: Finding the Invisible Wall

You can't see these issues with standard access logs because the logs only show total time. They don't show waiting time.

You need a sampling profiler. The best tool for Python is py-spy. It reads memory directly, so it has almost zero overhead and works on production servers.

1. Install py-spy:

Bash

pip install py-spy

2. Peek at the running process: Find your uvicorn Process ID (PID) and run:

Bash

# --rate 100: samples 100 times per second
# --subprocesses: checks worker processes too
sudo py-spy top --pid <PID> --rate 100 --subprocesses

3. Interpret the Output:

If you see time.sleep, select, or requests at the top of the list: You are blocking the loop.
If you see _run_until_complete dominating without specific function calls: You might be starving the loop with too many tasks.

Conclusion

Low CPU + High Latency is almost always a symptom of a Blocked Event Loop.

FastAPI apps don’t decay randomly.

They slow down because something is blocking the event loop.

Once you understand resource ownership, what runs on the loop, what must not, and what belongs in the application lifespan performance issues stop being mysterious.

If you’re also seeing database pool exhaustion or connection timeouts, read my guide on Fixing QueuePool Overflow Errors in FastAPI, where lifecycle bugs surface again in a different form.

To keep your FastAPI app responsive:

Audit your async def endpoints: Ensure no requests, time.sleep, or heavy math exists there.
Use asyncio.to_thread: For anything blocking or CPU-intensive (like password hashing).
Reuse HTTP Clients: Treat httpx.AsyncClient like a database connection - create it once, use it everywhere.

If you are struggling with database connection pool limits (another common cause of latency), read my guide on Fixing QueuePool Overflow errors.

Welcome to LogicLoopTech

Ayush Kumar — Tue, 14 Jul 2026 13:55:13 +0000

LogicLoopTech is a backend engineering blog. The focus is practical, hands-on content for developers building real systems with Python, FastAPI, AI, and RAG.

What you'll find here

How AI tooling actually helps backend developers - not the marketing version
RAG architectures that work in production, not just in demos
FastAPI patterns for shipping reliable APIs
Debugging guides for the kinds of bugs that eat your afternoon
Honest takes on what's worth using and what isn't

Why this exists

Most backend content is either too shallow ("here's a hello world") or too academic. The goal here is the middle ground: enough depth to actually use, short enough to read in one sitting.

New articles drop every two days. Let's go.