Build a FastAPI Chatbot with LangChain and WebSocket – Step‑by‑Step Guide

Prerequisites

Install the required Python packages listed in requirements.txt:

fastapi==0.108.0
langchain_core==0.1.28
langchain_openai==0.0.5
langchain_community==0.0.25
langchain==0.1.10
redis==7.2.0
qdrant_client==1.7.1
uvicorn==0.23.2

Install them with: pip install -r requirements.txt Check a dependency, e.g.:

pip show fastapi

Basic FastAPI Application

Import FastAPI, create an app instance, define a root route, and run with uvicorn:

# This is a simple FastAPI example
from fastapi import FastAPI

app = FastAPI()

@app.get("/")
def read_root():
    return {"Hello": "World"}

if __name__ == "__main__":
    import uvicorn
    uvicorn.run(app, host="localhost", port=8090)

Adding a Simple Chat Endpoint

Define a POST route that returns a static response:

@app.post("/chat")
def chat():
    return {"response": "I am a chat bot!"}

Integrating LangChain for AI‑Powered Chat

Import LangChain components, configure the OpenAI‑compatible model, define a tool, and build an agent executor inside a Master class:

import os
from dotenv import load_dotenv, find_dotenv
from fastapi import FastAPI, WebSocket, WebSocketDisconnect, BackgroundTasks
from langchain_openai import ChatOpenAI
from langchain.agents import create_openai_tools_agent, AgentExecutor, tool
from langchain.prompts import ChatPromptTemplate, MessagesPlaceholder

load_dotenv(find_dotenv())
os.environ["DASHSCOPE_API_KEY"] = "xxx"

class Master:
    def __init__(self):
        self.chatmodel = ChatOpenAI(
            api_key=os.getenv("DASHSCOPE_API_KEY"),
            base_url="https://dashscope.aliyuncs.com/compatible-mode/v1",
            model="qwen-plus",
            temperature=0,
            streaming=True,
        )
        self.prompt = ChatPromptTemplate.from_messages([
            ("system", "你是一个助手"),
            ("user", "{input}"),
            MessagesPlaceholder(variable_name="agent_scratchpad"),
        ])
        tools = [test]
        agent = create_openai_tools_agent(self.chatmodel, tools=tools, prompt=self.prompt)
        self.agent_executor = AgentExecutor(agent=agent, tools=tools, verbose=True)

    def run(self, query):
        result = self.agent_executor.invoke({"input": query})
        return result

Extended API Routes

PDF and text upload endpoints simply return confirmation messages:

@app.post("/add_pdfs")
def add_pdfs():
    return {"response": "PDFs added!"}

@app.post("add_texts")
def add_texts():
    return {"response": "Texts added!"}

WebSocket Echo Service

Implement a WebSocket endpoint that echoes received messages and handles disconnects:

@app.websocket("/ws")
async def websocket_endpoint(websocket: WebSocket):
    await websocket.accept()
    try:
        while True:
            data = await websocket.receive_text()
            await websocket.send_text(f"Message text was: {data}")
    except WebSocketDisconnect:
        print("Connection closed")
        await websocket.close()

Running the Server

Start the application with uvicorn:

uvicorn.run(app, host="localhost", port=8090)

Testing with Postman and Browser

Use Postman to send HTTP requests to the root, chat, PDF, and text endpoints, and to open the WebSocket connection. Screenshots of the requests and responses are shown below: