Merge pull request langchain-ai#240 from Sema4AI/introduce-postgres

Introduce postgres

.env.example

@@ -9,3 +9,7 @@ AZURE_OPENAI_API_VERSION=placeholder
 CONNERY_RUNNER_URL=https://your-personal-connery-runner-url
 CONNERY_RUNNER_API_KEY=placeholder
 PROXY_URL=your_proxy_url
+POSTGRES_PORT=placeholder
+POSTGRES_DB=placeholder
+POSTGRES_USER=placeholder
+POSTGRES_PASSWORD=placeholder

.env.gcp.yaml.example

@@ -2,7 +2,6 @@ OPENAI_API_KEY: your_secret_key_here
 LANGCHAIN_TRACING_V2: "true"
 LANGCHAIN_PROJECT: langserve-launch-example
 LANGCHAIN_API_KEY: your_secret_key_here
-REDIS_URL: your_secret_here
 FIREWORKS_API_KEY: your_secret_here
 AWS_ACCESS_KEY_ID: your_secret_here
 AWS_SECRET_ACCESS_KEY: your_secret_here
@@ -11,3 +10,10 @@ AZURE_OPENAI_API_BASE: your_secret_here
 AZURE_OPENAI_API_VERSION: your_secret_here
 AZURE_OPENAI_API_KEY: your_secret_here
 KAY_API_KEY: your_secret_here
+CONNERY_RUNNER_URL: https://your-personal-connery-runner-url
+CONNERY_RUNNER_API_KEY: your_secret_here
+POSTGRES_HOST: your_postgres_host_here
+POSTGRES_PORT: your_postgres_port_here
+POSTGRES_DB: your_postgres_db_here
+POSTGRES_USER: your_postgres_user_here
+POSTGRES_PASSWORD: your_postgres_password_here

.github/workflows/ci.yml

@@ -43,16 +43,20 @@ jobs:
     name: Python ${{ matrix.python-version }} tests
     services:
       # Label used to access the service container
-      redis:
-        image: redislabs/redisearch:latest
-        # Set health checks to wait until redis has started
+      postgres:
+        image: pgvector/pgvector:pg16
+        env:
+          POSTGRES_USER: postgres
+          POSTGRES_PASSWORD: postgres
+          POSTGRES_DB: postgres
+        # Set health checks to wait until postgres has started
         options: >-
-          --health-cmd "redis-cli ping"
+          --health-cmd pg_isready
           --health-interval 10s
           --health-timeout 5s
           --health-retries 5
         ports:
-          - "6379:6379"
+          - "5432:5432"
     steps:
       - uses: actions/checkout@v3
       - name: Set up Python ${{ matrix.python-version }} + Poetry ${{ env.POETRY_VERSION }}
@@ -62,12 +66,21 @@ jobs:
           poetry-version: ${{ env.POETRY_VERSION }}
           working-directory: .
           cache-key: langserve-all
-      - name: install redis
+      - name: Install dependencies
         run: |
-          pip install redis
-      - name: check redis is running
+          poetry install --with test
+      - name: Install golang-migrate
         run: |
-          python -c "import redis; redis.Redis(host='localhost', port=6379).ping()"
+          wget -O golang-migrate.deb https://github.com/golang-migrate/migrate/releases/download/v4.17.0/migrate.linux-amd64.deb
+          sudo dpkg -i golang-migrate.deb && rm golang-migrate.deb
+      - name: Run tests
+        env:
+          POSTGRES_HOST: localhost
+          POSTGRES_PORT: 5432
+          POSTGRES_DB: postgres
+          POSTGRES_USER: postgres
+          POSTGRES_PASSWORD: postgres
+        run: make test
 
   frontend-lint-and-build:
       runs-on: ubuntu-latest

.gitignore

@@ -1,6 +1,6 @@
 *.env
 .env.gcp.yaml
-redis-volume/
+postgres-volume/
 backend/ui
 
 # Operating System generated files

Dockerfile

@@ -17,6 +17,9 @@ FROM python:3.11
 
 # Install system dependencies
 RUN apt-get update && apt-get install -y libmagic1 && rm -rf /var/lib/apt/lists/*
+RUN wget -O golang-migrate.deb https://github.com/golang-migrate/migrate/releases/download/v4.17.0/migrate.linux-amd64.deb \
+    && dpkg -i golang-migrate.deb \
+    && rm golang-migrate.deb
 
 # Install Poetry
 RUN pip install poetry

README.md

@@ -48,19 +48,24 @@ poetry install
 
 **Set up persistence layer**
 
-The backend by default uses Redis for saving agent configurations and chat message history.
-In order to you use this, you need to a `REDIS_URL` variable.
+The backend uses Postgres for saving agent configurations and chat message history.
+In order to use this, you need to set the following environment variables:
 
 ```shell
-export REDIS_URL=...
+export POSTGRES_HOST=...
+export POSTGRES_PORT=...
+export POSTGRES_DB=...
+export POSTGRES_USER=...
+export POSTGRES_PASSWORD=...
 ```
 
+Migrations are managed with [golang-migrate](https://github.com/golang-migrate/migrate). Ensure it's installed on your machine and use `make migrate` to run all migrations. For those opting to run the project via Docker, the Dockerfile already includes golang-migrate.
+
 **Set up vector database**
 
-The backend by default also uses Redis as a vector database,
+The backend by default also uses Postgres as a vector database,
 although you can easily switch this out to use any of the 50+ vector databases in LangChain.
-If you are using Redis as a vectorstore, the above environment variable should work
-(assuming you've enabled `redissearch`)
+If you are using Postgres as a vectorstore, the above environment variables should work.
 
 **Set up language models**
 
@@ -112,7 +117,7 @@ Navigate to [http://localhost:5173/](http://localhost:5173/) and enjoy!
 
 ## Installation and Running with Docker
 
-This project supports a Docker-based setup, streamlining installation and execution. It automatically builds images for the frontend and backend and sets up Redis using docker-compose.
+This project supports a Docker-based setup, streamlining installation and execution. It automatically builds images for the frontend and backend and sets up Postgres using docker-compose.
 
 ### Quick Start
 
@@ -131,7 +136,7 @@ This project supports a Docker-based setup, streamlining installation and execut
    docker compose up
    ```
 
-   This command builds the Docker images for the frontend and backend from their respective Dockerfiles and starts all necessary services, including Redis.
+   This command builds the Docker images for the frontend and backend from their respective Dockerfiles and starts all necessary services, including Postgres.
 
 3. **Access the Application:**  
    With the services running, access the frontend at [http://localhost:5173](http://localhost:5173), substituting `5173` with the designated port number.

backend/Dockerfile

@@ -3,6 +3,9 @@ FROM python:3.11
 
 # Install system dependencies
 RUN apt-get update && apt-get install -y libmagic1 && rm -rf /var/lib/apt/lists/*
+RUN wget -O golang-migrate.deb https://github.com/golang-migrate/migrate/releases/download/v4.17.0/migrate.linux-amd64.deb \
+    && dpkg -i golang-migrate.deb \
+    && rm golang-migrate.deb
 
 # Install Poetry
 RUN pip install poetry
@@ -13,10 +16,9 @@ WORKDIR /backend
 # Copy only dependencies
 COPY pyproject.toml poetry.lock* ./
 
-# Install dependencies
-# --only main: Skip installing packages listed in the [tool.poetry.dev-dependencies] section
+# Install all dependencies
 RUN poetry config virtualenvs.create false \
-    && poetry install --no-interaction --no-ansi --only main
+    && poetry install --no-interaction --no-ansi
 
 # Copy the rest of application code
 COPY . .

backend/Makefile

@@ -16,6 +16,9 @@ TEST_FILE ?= tests/unit_tests/
 start:
 	poetry run uvicorn app.server:app --reload --port 8100
 
+migrate:
+	migrate -database postgres://$(POSTGRES_USER):$(POSTGRES_PASSWORD)@$(POSTGRES_HOST):$(POSTGRES_PORT)/$(POSTGRES_DB)?sslmode=disable -path ./migrations up
+
 test:
 	# We need to update handling of env variables for tests
 	YDC_API_KEY=placeholder OPENAI_API_KEY=placeholder poetry run pytest $(TEST_FILE)

backend/app/agent.py

@@ -12,7 +12,7 @@
 from app.agent_types.openai_agent import get_openai_agent_executor
 from app.agent_types.xml_agent import get_xml_agent_executor
 from app.chatbot import get_chatbot_executor
-from app.checkpoint import RedisCheckpoint
+from app.checkpoint import PostgresCheckpoint
 from app.llms import (
     get_anthropic_llm,
     get_google_llm,
@@ -67,7 +67,7 @@ class AgentType(str, Enum):
 
 DEFAULT_SYSTEM_MESSAGE = "You are a helpful assistant."
 
-CHECKPOINTER = RedisCheckpoint(at=CheckpointAt.END_OF_STEP)
+CHECKPOINTER = PostgresCheckpoint(at=CheckpointAt.END_OF_STEP)
 
 
 def get_agent_executor(

backend/app/api/assistants.py

@@ -5,7 +5,7 @@
 from pydantic import BaseModel, Field
 
 import app.storage as storage
-from app.schema import Assistant, AssistantWithoutUserId, OpengptsUserId
+from app.schema import Assistant, OpengptsUserId
 
 router = APIRouter()
 
@@ -24,42 +24,42 @@ class AssistantPayload(BaseModel):
 
 
 @router.get("/")
-def list_assistants(opengpts_user_id: OpengptsUserId) -> List[AssistantWithoutUserId]:
+async def list_assistants(opengpts_user_id: OpengptsUserId) -> List[Assistant]:
     """List all assistants for the current user."""
-    return storage.list_assistants(opengpts_user_id)
+    return await storage.list_assistants(opengpts_user_id)
 
 
 @router.get("/public/")
-def list_public_assistants(
+async def list_public_assistants(
     shared_id: Annotated[
         Optional[str], Query(description="ID of a publicly shared assistant.")
     ] = None,
-) -> List[AssistantWithoutUserId]:
+) -> List[Assistant]:
     """List all public assistants."""
-    return storage.list_public_assistants(
+    return await storage.list_public_assistants(
         FEATURED_PUBLIC_ASSISTANTS + ([shared_id] if shared_id else [])
     )
 
 
 @router.get("/{aid}")
-def get_asistant(
+async def get_assistant(
     opengpts_user_id: OpengptsUserId,
     aid: AssistantID,
 ) -> Assistant:
     """Get an assistant by ID."""
-    assistant = storage.get_assistant(opengpts_user_id, aid)
+    assistant = await storage.get_assistant(opengpts_user_id, aid)
     if not assistant:
         raise HTTPException(status_code=404, detail="Assistant not found")
     return assistant
 
 
 @router.post("")
-def create_assistant(
+async def create_assistant(
     opengpts_user_id: OpengptsUserId,
     payload: AssistantPayload,
 ) -> Assistant:
     """Create an assistant."""
-    return storage.put_assistant(
+    return await storage.put_assistant(
         opengpts_user_id,
         str(uuid4()),
         name=payload.name,
@@ -69,13 +69,13 @@ def create_assistant(
 
 
 @router.put("/{aid}")
-def upsert_assistant(
+async def upsert_assistant(
     opengpts_user_id: OpengptsUserId,
     aid: AssistantID,
     payload: AssistantPayload,
 ) -> Assistant:
     """Create or update an assistant."""
-    return storage.put_assistant(
+    return await storage.put_assistant(
         opengpts_user_id,
         aid,
         name=payload.name,

backend/app/api/runs.py

@@ -1,4 +1,3 @@
-import asyncio
 import json
 from typing import Optional, Sequence
 
@@ -16,7 +15,7 @@
 
 from app.agent import agent
 from app.schema import OpengptsUserId
-from app.storage import get_assistant, public_user_id
+from app.storage import get_assistant
 from app.stream import astream_messages, to_sse
 
 router = APIRouter()
@@ -35,15 +34,7 @@ async def _run_input_and_config(request: Request, opengpts_user_id: OpengptsUser
         body = await request.json()
     except json.JSONDecodeError:
         raise RequestValidationError(errors=["Invalid JSON body"])
-    assistant, public_assistant = await asyncio.gather(
-        asyncio.get_running_loop().run_in_executor(
-            None, get_assistant, opengpts_user_id, body["assistant_id"]
-        ),
-        asyncio.get_running_loop().run_in_executor(
-            None, get_assistant, public_user_id, body["assistant_id"]
-        ),
-    )
-    assistant = assistant or public_assistant
+    assistant = await get_assistant(opengpts_user_id, body["assistant_id"])
     if not assistant:
         raise HTTPException(status_code=404, detail="Assistant not found")
     config: RunnableConfig = {

backend/app/api/threads.py

@@ -6,7 +6,7 @@
 from pydantic import BaseModel, Field
 
 import app.storage as storage
-from app.schema import OpengptsUserId, Thread, ThreadWithoutUserId
+from app.schema import OpengptsUserId, Thread
 
 router = APIRouter()
 
@@ -28,49 +28,49 @@ class ThreadMessagesPostRequest(BaseModel):
 
 
 @router.get("/")
-def list_threads(opengpts_user_id: OpengptsUserId) -> List[ThreadWithoutUserId]:
+async def list_threads(opengpts_user_id: OpengptsUserId) -> List[Thread]:
     """List all threads for the current user."""
-    return storage.list_threads(opengpts_user_id)
+    return await storage.list_threads(opengpts_user_id)
 
 
 @router.get("/{tid}/messages")
-def get_thread_messages(
+async def get_thread_messages(
     opengpts_user_id: OpengptsUserId,
     tid: ThreadID,
 ):
     """Get all messages for a thread."""
-    return storage.get_thread_messages(opengpts_user_id, tid)
+    return await storage.get_thread_messages(opengpts_user_id, tid)
 
 
 @router.post("/{tid}/messages")
-def add_thread_messages(
+async def add_thread_messages(
     opengpts_user_id: OpengptsUserId,
     tid: ThreadID,
     payload: ThreadMessagesPostRequest,
 ):
     """Add messages to a thread."""
-    return storage.post_thread_messages(opengpts_user_id, tid, payload.messages)
+    return await storage.post_thread_messages(opengpts_user_id, tid, payload.messages)
 
 
 @router.get("/{tid}")
-def get_thread(
+async def get_thread(
     opengpts_user_id: OpengptsUserId,
     tid: ThreadID,
 ) -> Thread:
     """Get a thread by ID."""
-    thread = storage.get_thread(opengpts_user_id, tid)
+    thread = await storage.get_thread(opengpts_user_id, tid)
     if not thread:
         raise HTTPException(status_code=404, detail="Thread not found")
     return thread
 
 
 @router.post("")
-def create_thread(
+async def create_thread(
     opengpts_user_id: OpengptsUserId,
     thread_put_request: ThreadPutRequest,
 ) -> Thread:
     """Create a thread."""
-    return storage.put_thread(
+    return await storage.put_thread(
         opengpts_user_id,
         str(uuid4()),
         assistant_id=thread_put_request.assistant_id,
@@ -79,13 +79,13 @@ def create_thread(
 
 
 @router.put("/{tid}")
-def upsert_thread(
+async def upsert_thread(
     opengpts_user_id: OpengptsUserId,
     tid: ThreadID,
     thread_put_request: ThreadPutRequest,
 ) -> Thread:
     """Update a thread."""
-    return storage.put_thread(
+    return await storage.put_thread(
         opengpts_user_id,
         tid,
         assistant_id=thread_put_request.assistant_id,