Python FastAPI state not exist on Vercel but does locally - lifespan

This follows on from Python typings packages breaks FastAPI on Vercel - #3 by brooke-tintunacom.

After I fixed that problem (using python package for Typing) and solving some other problems I now have a problem where the State I’m storing does not exist when the app runs on Vercel, however it works fine when running locally. Any ideas?

The code is:

$ cat requirements.txt
python-dotenv
openai
instructor
rich
pydantic
fastapi
mangum
uvicorn
atomic-agents
$ cat vercel.json
{
    "version": 2,
    "builds": [
     {
      "src": "app/main.py",
      "use": "@vercel/python"
     }
    ],
    "routes": [
     {
      "src": "/(.*)",
      "dest": "app/main.py"
     }
    ]
   }

$ cat app/lifespan.py
from contextlib import asynccontextmanager
from app.llm.atomic_agents_service import AtomicAgentsService

@asynccontextmanager
async def lifespan(app):
    # Load the LLM service - do it this way so that it is loaded once and not on every request
    atomic_agents_service = AtomicAgentsService()
    print(" - created atomic_agents_service: ", atomic_agents_service)
    app.state.atomic_agents_service = atomic_agents_service
    yield
    # Clean up the LLM service and release the resources
    app.state.atomic_agents_service = None
$ cat app/main.py
from fastapi import FastAPI
from fastapi.middleware.cors import CORSMiddleware
from mangum import Mangum
from .routes import router as api_router
from app.lifespan import lifespan

app = FastAPI(lifespan=lifespan)

# CORS configuration
app.add_middleware(
    CORSMiddleware,
    allow_origins=["*"],  # Allows all origins
    allow_credentials=True,
    allow_methods=["*"],
    allow_headers=["*"],
)

app.include_router(api_router)

@app.get("/")
async def read_root():
    return {"health": "up"}
$ cat app/routes.py
from app.atomic_agents_data import AtomicAgentMemoryModel
from app.llm.atomic_agents_service import AtomicAgentsService
from rich.console import Console
import json

router = APIRouter()

console = Console()

@router.post("/choice")
async def post_choice(request: Request, choice: str = None, memory: AtomicAgentMemoryModel = None):
    console.print("/choice")
    console.print(" - choice: ", choice)
    console.print(" - memory: ", memory.json()) if memory else console.print(" - memory: None")
    console.print("IN BETWEEN")
    console.print(" - request.app.state: ", request.app.state)
    console.print(" - request.app.state dir: ", dir(request.app.state))
    console.print(" - request.app.state state: ", request.app.state._state)
    console.print(" - atomic_agents_service: ", request.app.state.atomic_agents_service)
    # `memory` will automatically be deserialized into an AtomicAgentMemoryModel instance
    if (memory is None or memory.history is None or len(memory.history) == 0):
        response = request.app.state.atomic_agents_service.start()
    else:
        response = request.app.state.atomic_agents_service.post_choice(choice, memory)

    # Perform your logic here
    return {"request": choice, "response": response}

Locally the terminal shows:

AtomicAgentsService - init
 - created atomic_agents_service:  <app.llm.atomic_agents_service.AtomicAgentsService object at 0x1087fd940>
...
/choice
 - choice:  None
 - memory: None
IN BETWEEN
 - request.app.state:  <starlette.datastructures.State object at 0x1087cf0e0>
 - request.app.state state:
{'atomic_agents_service': <app.llm.atomic_agents_service.AtomicAgentsService object at 0x1087fd940>}
 - atomic_agents_service:  <app.llm.atomic_agents_service.AtomicAgentsService object at 0x1087fd940>

But the vercel runtime log shows (just request and message):

# NOTE --> " - created atomic_agents_service:  ..." is misssing <-- NOTE
/choice    127.0.0.1 - - [30/Nov/2024 04:44:52] "POST /choice HTTP/1.1" 500 -
/choice.   - request.app.state state: {}
/choice.  - request.app.state: <starlette.datastructures.State object at 0x7f5d466da810>
/choice.   IN BETWEEN
/choice    - memory: None
/choice.   - choice: None

It looks like the lifespan is not executing. There is no - atomic_agents_service: xxxxx in the Vercel runtime log.

Any ideas?

Ok, after a bit of back-and-forth with the Vercel AI, we came up with the following solution. It is untested but from my limited experience it should work - with two gotchas. 1) I’ve had problems with Depends() before; 2) events in FastAPI are depreciated. But I’ll worry about that if I get it working (and committed):

I’ll let this thread know if this works.

BTW the AI is pretty good - it would be better if it wasn’t stuck in a tiny window.


Thank you for clarifying the requirements. I’ll create a solution that meets all three of these requirements without using Mangum. Let’s go through the implementation step by step:

Let’s create the files as per your requirements:

  1. First, let’s create a separate file for your service:
# atomic_agents_service.py
class AtomicAgentsService:
    def __init__(self):
        # Initialize your service here
        self.loaded = False

    def load(self):
        # Load the service
        self.loaded = True

    def process(self, data):
        if not self.loaded:
            raise RuntimeError("Service not loaded")
        # Your processing logic here
        return f"Processed: {data}"

    def unload(self):
        # Clean up resources
        self.loaded = False
  1. Now, let’s create a routes.py file:
# routes.py
from fastapi import APIRouter, Depends, HTTPException
from atomic_agents_service import AtomicAgentsService

router = APIRouter()

# Dependency
def get_atomic_agents_service():
    return router.app.state.atomic_agents_service

@router.get("/process")
async def process(data: str, service: AtomicAgentsService = Depends(get_atomic_agents_service)):
    try:
        result = service.process(data)
        return {"result": result}
    except RuntimeError as e:
        raise HTTPException(status_code=500, detail=str(e))

@router.get("/health")
async def health_check(service: AtomicAgentsService = Depends(get_atomic_agents_service)):
    if not service.loaded:
        raise HTTPException(status_code=500, detail="Service not loaded")
    return {"status": "ok"}

# Add more endpoints as needed
  1. Now, let’s create the main app.py file:
# app.py
from fastapi import FastAPI
from atomic_agents_service import AtomicAgentsService
from routes import router as api_router

app = FastAPI()

@app.on_event("startup")
async def startup_event():
    app.state.atomic_agents_service = AtomicAgentsService()
    app.state.atomic_agents_service.load()

@app.on_event("shutdown")
async def shutdown_event():
    app.state.atomic_agents_service.unload()

app.include_router(api_router)
  1. Create a vercel.json file to configure your Vercel deployment:
{
  "version": 2,
  "builds": [
    {
      "src": "app.py",
      "use": "@vercel/python",
      "config": {
        "maxLambdaSize": "50mb"
      }
    }
  ],
  "routes": [
    {
      "src": "/(.*)",
      "dest": "/app.py"
    }
  ]
}

Key points and explanations:

  1. We’ve separated the service (AtomicAgentsService) into its own module (atomic_agents_service.py).
  2. The routes.py file contains all your API endpoints and uses FastAPI’s dependency injection to access the service.
  3. The app.py file manages the FastAPI app, including the startup and shutdown events to initialize and clean up the AtomicAgentsService.
  4. We’re using FastAPI’s app.state to store the service instance, which allows sharing between different modules.
  5. We’ve added error handling in the routes to manage cases where the service might not be loaded.

This setup meets all your requirements:

  1. It separates main.py (now app.py) and routes.py for API setup and handling.
  2. It creates the service upon FastAPI startup and shares it between the two files using app.state.
  3. It doesn’t use Mangum or any other ASGI adapter, as it’s not needed for Vercel deployments.

To run your app locally for testing, you can use Uvicorn:

uvicorn app:app --reload

This approach ensures that your AtomicAgentsService is initialized when the FastAPI app starts, and it’s accessible in your route handlers through FastAPI’s dependency injection system. The service is properly loaded and unloaded during the application lifecycle, and it’s shared between the main app and the routes module.

Remember to test your application locally before deploying to Vercel. This setup should work correctly both locally and when deployed to Vercel, without the need for Mangum or any other ASGI adapter.

1 Like

Ok that suggestion didn’t quite work and I had to iterate quite a number of times more with Vercel AI. Here is the result for the 2 files different to the above.

I don’t like this version as it seems inelegant, relies on globals and the shutdown needs to be manually performed through an API endpoint.

Does anyone have suggestions for a better way of writing this?

# routes.py
from fastapi import APIRouter, Depends, HTTPException, Request

from app.atomic_agents_data import AtomicAgentMemoryModel
from app.llm.atomic_agents_service import AtomicAgentsService
from rich.console import Console

router = APIRouter()

console = Console()

def get_atomic_agents_service(request: Request):
    console.print(" - gaas request.app: ", request.app)
    console.print(" - gaas request.app.state state: ", request.app.state)
    console.print(" - gaas request.app.state state: ", request.app.state._state)
    return request.app.state.atomic_agents_service

@router.post("/choice")
async def post_choice(request: Request, choice: str = None, memory: AtomicAgentMemoryModel = None, service: AtomicAgentsService = Depends(get_atomic_agents_service)):
    console.print("/choice")
    console.print(" - choice: ", choice)
    console.print(" - memory: ", memory.json()) if memory else console.print(" - memory: None")
    console.print("IN BETWEEN")
    console.print(" - post_choice request.app.state state: ", request.app.state)
    console.print(" - post_choice request.app.state state: ", request.app.state._state)

    console.print(" - atomic_agents_service: ", service)
    ...
# main.py
rom fastapi import FastAPI
from fastapi.middleware.cors import CORSMiddleware
from rich.console import Console

from app.llm.atomic_agents_service import AtomicAgentsService
from .routes import router as api_router

console = Console()

app = FastAPI()

# CORS configuration
app.add_middleware(
    CORSMiddleware,
    allow_origins=["*"],  # Allows all origins
    allow_credentials=True,
    allow_methods=["*"],
    allow_headers=["*"],
)

def initialize_service():
    service = AtomicAgentsService()
    # service.load()
    return service

# Initialize the service and set it in FastAPI's state
app.state.atomic_agents_service = initialize_service()
console.print("starting up FastAPI - atomic_agents_service: ", app.state.atomic_agents_service)

# Have to manually cause shutdown
async def shutdown():
    if app.state.atomic_agents_service:
        # app.state.atomic_agents_service.unload()
        app.state.atomic_agents_service = None
    return {"message": "Service cleaned up"}

@app.get("/")
async def read_root():
    console.print("health is up")
    return {"health": "up"}

app.include_router(api_router)
1 Like

Hi, Brooke! Thanks for coming back and sharing your insights :smile:

I understand your concerns about the current implementation! I have a few suggestions.

  1. First, let’s update the main.py file:
# main.py
from fastapi import FastAPI
from fastapi.middleware.cors import CORSMiddleware
from contextlib import asynccontextmanager
from app.llm.atomic_agents_service import AtomicAgentsService
from .routes import router as api_router

@asynccontextmanager
async def lifespan(app: FastAPI):
    # Startup: create and load the service
    app.state.atomic_agents_service = AtomicAgentsService()
    # If you need to load the service, uncomment the next line
    # await app.state.atomic_agents_service.load()
    yield
    # Shutdown: clean up the service
    # If you need to unload the service, uncomment the next line
    # await app.state.atomic_agents_service.unload()
    app.state.atomic_agents_service = None

app = FastAPI(lifespan=lifespan)

# CORS configuration
app.add_middleware(
    CORSMiddleware,
    allow_origins=["*"],
    allow_credentials=True,
    allow_methods=["*"],
    allow_headers=["*"],
)

app.include_router(api_router)

@app.get("/")
async def read_root():
    return {"health": "up"}
  1. Now, let’s update the routes.py file:
# routes.py
from fastapi import APIRouter, Depends, HTTPException, Request
from app.atomic_agents_data import AtomicAgentMemoryModel
from app.llm.atomic_agents_service import AtomicAgentsService
from rich.console import Console

router = APIRouter()
console = Console()

def get_atomic_agents_service(request: Request) -> AtomicAgentsService:
    service = request.app.state.atomic_agents_service
    if service is None:
        raise HTTPException(status_code=503, detail="Service not initialized")
    return service

@router.post("/choice")
async def post_choice(
    request: Request,
    choice: str = None,
    memory: AtomicAgentMemoryModel = None,
    service: AtomicAgentsService = Depends(get_atomic_agents_service)
):
    console.print("/choice")
    console.print(" - choice: ", choice)
    console.print(" - memory: ", memory.json() if memory else "None")
    console.print(" - atomic_agents_service: ", service)

    # Your existing logic here
    # ...

    return {"request": choice, "response": "Your response here"}

This implementation does the following:

  1. It uses FastAPI’s lifespan feature, which is the recommended way to handle startup and shutdown events. This replaces the need for @app.on_event("startup") and @app.on_event("shutdown") decorators.
  2. The AtomicAgentsService is created when the application starts and is automatically cleaned up when the application shuts down.
  3. We’ve removed the global variables and manual initialization.
  4. The get_atomic_agents_service dependency function now raises an HTTP exception if the service is not initialized, providing better error handling.
  5. We’re using type hints more consistently, which improves code readability and helps catch potential errors.
  6. The shutdown process is handled automatically by the lifespan context manager, eliminating the need for a manual API endpoint to perform shutdown.

BTW I got these pointers from v0.dev - might be worth checking out in case you haven’t already (unsure if that’s what you meant with “Vercel AI”)