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

(edited)

This follows on from https://community.vercel.com/t/python-typings-packages-breaks-fastapi-on-vercel/2598/3.

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:

1$ cat requirements.txt2python-dotenv3openai4instructor5rich6pydantic7fastapi8mangum9uvicorn10atomic-agents

1$ cat vercel.json2{3    "version": 2,4    "builds": [5     {6      "src": "app/main.py",7      "use": "@vercel/python"8     }9    ],10    "routes": [11     {12      "src": "/(.*)",13      "dest": "app/main.py"14     }15    ]16   }17

1$ cat app/lifespan.py2from contextlib import asynccontextmanager3from app.llm.atomic_agents_service import AtomicAgentsService45@asynccontextmanager6async def lifespan(app):7    # Load the LLM service - do it this way so that it is loaded once and not on every request8    atomic_agents_service = AtomicAgentsService()9    print(" - created atomic_agents_service: ", atomic_agents_service)10    app.state.atomic_agents_service = atomic_agents_service11    yield12    # Clean up the LLM service and release the resources13    app.state.atomic_agents_service = None

1$ cat app/main.py2from fastapi import FastAPI3from fastapi.middleware.cors import CORSMiddleware4from mangum import Mangum5from .routes import router as api_router6from app.lifespan import lifespan78app = FastAPI(lifespan=lifespan)910# CORS configuration11app.add_middleware(12    CORSMiddleware,13    allow_origins=["*"],  # Allows all origins14    allow_credentials=True,15    allow_methods=["*"],16    allow_headers=["*"],17)1819app.include_router(api_router)2021@app.get("/")22async def read_root():23    return {"health": "up"}

1$ cat app/routes.py2from app.atomic_agents_data import AtomicAgentMemoryModel3from app.llm.atomic_agents_service import AtomicAgentsService4from rich.console import Console5import json67router = APIRouter()89console = Console()1011@router.post("/choice")12async def post_choice(request: Request, choice: str = None, memory: AtomicAgentMemoryModel = None):13    console.print("/choice")14    console.print(" - choice: ", choice)15    console.print(" - memory: ", memory.json()) if memory else console.print(" - memory: None")16    console.print("IN BETWEEN")17    console.print(" - request.app.state: ", request.app.state)18    console.print(" - request.app.state dir: ", dir(request.app.state))19    console.print(" - request.app.state state: ", request.app.state._state)20    console.print(" - atomic_agents_service: ", request.app.state.atomic_agents_service)21    # `memory` will automatically be deserialized into an AtomicAgentMemoryModel instance22    if (memory is None or memory.history is None or len(memory.history) == 0):23        response = request.app.state.atomic_agents_service.start()24    else:25        response = request.app.state.atomic_agents_service.post_choice(choice, memory)2627    # Perform your logic here28    return {"request": choice, "response": response}

Locally the terminal shows:

1AtomicAgentsService - init2 - created atomic_agents_service:  <app.llm.atomic_agents_service.AtomicAgentsService object at 0x1087fd940>3...4/choice5 - choice:  None6 - memory: None7IN BETWEEN8 - request.app.state:  <starlette.datastructures.State object at 0x1087cf0e0>9 - request.app.state state:10{'atomic_agents_service': <app.llm.atomic_agents_service.AtomicAgentsService object at 0x1087fd940>}11 - atomic_agents_service:  <app.llm.atomic_agents_service.AtomicAgentsService object at 0x1087fd940>

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

1# NOTE --> " - created atomic_agents_service:  ..." is misssing <-- NOTE2/choice    127.0.0.1 - - [30/Nov/2024 04:44:52] "POST /choice HTTP/1.1" 500 -3/choice.   - request.app.state state: {}4/choice.  - request.app.state: <starlette.datastructures.State object at 0x7f5d466da810>5/choice.   IN BETWEEN6/choice    - memory: None7/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:

First, let's create a separate file for your service:

1# atomic_agents_service.py2class AtomicAgentsService:3    def __init__(self):4        # Initialize your service here5        self.loaded = False67    def load(self):8        # Load the service9        self.loaded = True1011    def process(self, data):12        if not self.loaded:13            raise RuntimeError("Service not loaded")14        # Your processing logic here15        return f"Processed: {data}"1617    def unload(self):18        # Clean up resources19        self.loaded = False

Now, let's create a routes.py file:

1# routes.py2from fastapi import APIRouter, Depends, HTTPException3from atomic_agents_service import AtomicAgentsService45router = APIRouter()67# Dependency8def get_atomic_agents_service():9    return router.app.state.atomic_agents_service1011@router.get("/process")12async def process(data: str, service: AtomicAgentsService = Depends(get_atomic_agents_service)):13    try:14        result = service.process(data)15        return {"result": result}16    except RuntimeError as e:17        raise HTTPException(status_code=500, detail=str(e))1819@router.get("/health")20async def health_check(service: AtomicAgentsService = Depends(get_atomic_agents_service)):21    if not service.loaded:22        raise HTTPException(status_code=500, detail="Service not loaded")23    return {"status": "ok"}2425# Add more endpoints as needed

Now, let's create the main app.py file:

1# app.py2from fastapi import FastAPI3from atomic_agents_service import AtomicAgentsService4from routes import router as api_router56app = FastAPI()78@app.on_event("startup")9async def startup_event():10    app.state.atomic_agents_service = AtomicAgentsService()11    app.state.atomic_agents_service.load()1213@app.on_event("shutdown")14async def shutdown_event():15    app.state.atomic_agents_service.unload()1617app.include_router(api_router)

Create a vercel.json file to configure your Vercel deployment:

1{2  "version": 2,3  "builds": [4    {5      "src": "app.py",6      "use": "@vercel/python",7      "config": {8        "maxLambdaSize": "50mb"9      }10    }11  ],12  "routes": [13    {14      "src": "/(.*)",15      "dest": "/app.py"16    }17  ]18}

Key points and explanations:

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

This setup meets all your requirements:

It separates main.py (now app.py) and routes.py for API setup and handling.
It creates the service upon FastAPI startup and shares it between the two files using app.state.
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:

1uvicorn 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.

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?

1# routes.py2from fastapi import APIRouter, Depends, HTTPException, Request34from app.atomic_agents_data import AtomicAgentMemoryModel5from app.llm.atomic_agents_service import AtomicAgentsService6from rich.console import Console78router = APIRouter()910console = Console()1112def get_atomic_agents_service(request: Request):13    console.print(" - gaas request.app: ", request.app)14    console.print(" - gaas request.app.state state: ", request.app.state)15    console.print(" - gaas request.app.state state: ", request.app.state._state)16    return request.app.state.atomic_agents_service1718@router.post("/choice")19async def post_choice(request: Request, choice: str = None, memory: AtomicAgentMemoryModel = None, service: AtomicAgentsService = Depends(get_atomic_agents_service)):20    console.print("/choice")21    console.print(" - choice: ", choice)22    console.print(" - memory: ", memory.json()) if memory else console.print(" - memory: None")23    console.print("IN BETWEEN")24    console.print(" - post_choice request.app.state state: ", request.app.state)25    console.print(" - post_choice request.app.state state: ", request.app.state._state)2627    console.print(" - atomic_agents_service: ", service)28    ...

1# main.py2rom fastapi import FastAPI3from fastapi.middleware.cors import CORSMiddleware4from rich.console import Console56from app.llm.atomic_agents_service import AtomicAgentsService7from .routes import router as api_router89console = Console()1011app = FastAPI()1213# CORS configuration14app.add_middleware(15    CORSMiddleware,16    allow_origins=["*"],  # Allows all origins17    allow_credentials=True,18    allow_methods=["*"],19    allow_headers=["*"],20)2122def initialize_service():23    service = AtomicAgentsService()24    # service.load()25    return service2627# Initialize the service and set it in FastAPI's state28app.state.atomic_agents_service = initialize_service()29console.print("starting up FastAPI - atomic_agents_service: ", app.state.atomic_agents_service)3031# Have to manually cause shutdown32async def shutdown():33    if app.state.atomic_agents_service:34        # app.state.atomic_agents_service.unload()35        app.state.atomic_agents_service = None36    return {"message": "Service cleaned up"}3738@app.get("/")39async def read_root():40    console.print("health is up")41    return {"health": "up"}4243app.include_router(api_router)

reply to #3

Vercel Alum

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

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

First, let's update the main.py file:

1# main.py2from fastapi import FastAPI3from fastapi.middleware.cors import CORSMiddleware4from contextlib import asynccontextmanager5from app.llm.atomic_agents_service import AtomicAgentsService6from .routes import router as api_router78@asynccontextmanager9async def lifespan(app: FastAPI):10    # Startup: create and load the service11    app.state.atomic_agents_service = AtomicAgentsService()12    # If you need to load the service, uncomment the next line13    # await app.state.atomic_agents_service.load()14    yield15    # Shutdown: clean up the service16    # If you need to unload the service, uncomment the next line17    # await app.state.atomic_agents_service.unload()18    app.state.atomic_agents_service = None1920app = FastAPI(lifespan=lifespan)2122# CORS configuration23app.add_middleware(24    CORSMiddleware,25    allow_origins=["*"],26    allow_credentials=True,27    allow_methods=["*"],28    allow_headers=["*"],29)3031app.include_router(api_router)3233@app.get("/")34async def read_root():35    return {"health": "up"}

Now, let's update the routes.py file:

1# routes.py2from fastapi import APIRouter, Depends, HTTPException, Request3from app.atomic_agents_data import AtomicAgentMemoryModel4from app.llm.atomic_agents_service import AtomicAgentsService5from rich.console import Console67router = APIRouter()8console = Console()910def get_atomic_agents_service(request: Request) -> AtomicAgentsService:11    service = request.app.state.atomic_agents_service12    if service is None:13        raise HTTPException(status_code=503, detail="Service not initialized")14    return service1516@router.post("/choice")17async def post_choice(18    request: Request,19    choice: str = None,20    memory: AtomicAgentMemoryModel = None,21    service: AtomicAgentsService = Depends(get_atomic_agents_service)22):23    console.print("/choice")24    console.print(" - choice: ", choice)25    console.print(" - memory: ", memory.json() if memory else "None")26    console.print(" - atomic_agents_service: ", service)2728    # Your existing logic here29    # ...3031    return {"request": choice, "response": "Your response here"}

This implementation does the following:

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.
The AtomicAgentsService is created when the application starts and is automatically cleaned up when the application shuts down.
We've removed the global variables and manual initialization.
The get_atomic_agents_service dependency function now raises an HTTP exception if the service is not initialized, providing better error handling.
We're using type hints more consistently, which improves code readability and helps catch potential errors.
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")

This topic is closed. It no longer accepts new replies.