LangServe : Déployer vos applications LangChain en APIs REST production-ready

from fastapi import FastAPI
from langchain_openai import ChatOpenAI

app = FastAPI()
llm = ChatOpenAI()

@app.post("/chat")
async def chat(request: dict):
    # Gérer la sérialisation
    # Gérer le streaming manuellement
    # Implémenter batch
    # Créer un playground
    # Gérer les erreurs
    # ... des heures de code boilerplate
    pass

from fastapi import FastAPI
from langserve import add_routes
from langchain_openai import ChatOpenAI

app = FastAPI()
llm = ChatOpenAI()

add_routes(app, llm, path="/chat")  # C'est tout !

# Installation de base
pip install langserve[all]

# Ou minimal
pip install langserve

# Pour le serveur
pip install "langserve[server]"

# Pour le client
pip install "langserve[client]"

from fastapi import FastAPI
from langserve import add_routes
from langchain_openai import ChatOpenAI

# 1. Créer l'app FastAPI
app = FastAPI(
    title="Mon API LangChain",
    version="1.0",
    description="API pour un chatbot IA"
)

# 2. Créer votre chain
llm = ChatOpenAI(model="gpt-4o-mini", temperature=0.7)

# 3. Ajouter la route (une seule ligne !)
add_routes(
    app,
    llm,
    path="/chat"
)

# 4. Lancer le serveur
if __name__ == "__main__":
    import uvicorn
    uvicorn.run(app, host="0.0.0.0", port=8000)

python server.py
# ou
uvicorn server:app --reload

from langserve import RemoteRunnable

# Connexion au serveur
chat = RemoteRunnable("http://localhost:8000/chat")

# Invoke
response = chat.invoke("Explique la photosynthèse")
print(response.content)

# Streaming
for chunk in chat.stream("Raconte une histoire"):
    print(chunk.content, end="", flush=True)

# Batch
responses = chat.batch([
    "Question 1",
    "Question 2",
    "Question 3"
])

from fastapi import FastAPI
from langserve import add_routes
from langchain_openai import ChatOpenAI, OpenAIEmbeddings
from langchain_community.vectorstores import FAISS
from langchain.chains import RetrievalQA

app = FastAPI(title="RAG API")

# Setup RAG
vectorstore = FAISS.load_local(
    "./vectorstore",
    OpenAIEmbeddings(),
    allow_dangerous_deserialization=True
)

llm = ChatOpenAI(model="gpt-4")

qa_chain = RetrievalQA.from_chain_type(
    llm=llm,
    retriever=vectorstore.as_retriever(search_kwargs={"k": 3}),
    return_source_documents=True
)

# Déployer
add_routes(
    app,
    qa_chain,
    path="/rag",
    enable_feedback_endpoint=True  # Active le feedback utilisateur
)

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

from langserve import RemoteRunnable

rag = RemoteRunnable("http://localhost:8000/rag")

result = rag.invoke({
    "query": "Qu'est-ce que le fine-tuning ?"
})

print(result["result"])
print("\nSources:")
for doc in result["source_documents"]:
    print(f"- {doc.metadata['source']}")

from fastapi import FastAPI
from langserve import add_routes
from langchain_openai import ChatOpenAI
from langchain.agents import AgentExecutor, create_react_agent
from langchain_community.tools import DuckDuckGoSearchRun
from langchain.tools import Tool

app = FastAPI(title="Agent API")

# Outils
search = DuckDuckGoSearchRun()
tools = [
    Tool(
        name="Search",
        func=search.run,
        description="Recherche sur le web pour informations actuelles"
    )
]

# Agent
llm = ChatOpenAI(model="gpt-4", temperature=0)
agent = create_react_agent(llm, tools, prompt)
agent_executor = AgentExecutor(
    agent=agent,
    tools=tools,
    verbose=True,
    max_iterations=5
)

# Déployer
add_routes(app, agent_executor, path="/agent")

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

from fastapi import FastAPI
from langserve import add_routes
from langchain_openai import ChatOpenAI
from langchain_core.prompts import ChatPromptTemplate
from langchain_core.output_parsers import StrOutputParser

app = FastAPI()

# Chain avec prompt template
prompt = ChatPromptTemplate.from_messages([
    ("system", "Tu es un expert en {domain}."),
    ("human", "{question}")
])

llm = ChatOpenAI(model="gpt-4o-mini")
output_parser = StrOutputParser()

chain = prompt | llm | output_parser

# Déployer
add_routes(app, chain, path="/expert")

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

from langserve import RemoteRunnable

expert = RemoteRunnable("http://localhost:8000/expert")

response = expert.invoke({
    "domain": "astronomie",
    "question": "Qu'est-ce qu'un trou noir ?"
})

print(response)

from fastapi import FastAPI
from langserve import add_routes
from langchain_openai import ChatOpenAI

app = FastAPI()

# LLM avec streaming activé
llm = ChatOpenAI(
    model="gpt-4",
    streaming=True,  # Important !
    temperature=0.7
)

add_routes(app, llm, path="/chat")

from langserve import RemoteRunnable

chat = RemoteRunnable("http://localhost:8000/chat")

# Streaming token par token
for chunk in chat.stream("Raconte une longue histoire"):
    print(chunk.content, end="", flush=True)

print()  # Nouvelle ligne

# Stream avec logs détaillés
for event in chat.stream_log("Question complexe"):
    # event contient : type, data, metadata
    if event["type"] == "chunk":
        print(event["data"]["content"], end="")

<!DOCTYPE html>
<html>
<head>
    <title>Chat Streaming</title>
</head>
<body>
    <textarea id="question" placeholder="Votre question..."></textarea>
    <button onclick="askQuestion()">Envoyer</button>
    <div id="response"></div>

    <script>
        async function askQuestion() {
            const question = document.getElementById('question').value;
            const responseDiv = document.getElementById('response');
            responseDiv.innerHTML = '';

            const response = await fetch('http://localhost:8000/chat/stream', {
                method: 'POST',
                headers: {'Content-Type': 'application/json'},
                body: JSON.stringify({input: question})
            });

            const reader = response.body.getReader();
            const decoder = new TextDecoder();

            while (true) {
                const {done, value} = await reader.read();
                if (done) break;

                const chunk = decoder.decode(value);
                const lines = chunk.split('\n');

                for (const line of lines) {
                    if (line.startsWith('data: ')) {
                        const data = JSON.parse(line.slice(6));
                        responseDiv.innerHTML += data.content;
                    }
                }
            }
        }
    </script>
</body>
</html>

from fastapi import FastAPI
from fastapi.middleware.cors import CORSMiddleware
from langserve import add_routes

app = FastAPI()

# Configuration CORS
app.add_middleware(
    CORSMiddleware,
    allow_origins=["http://localhost:3000"],  # Votre frontend
    allow_credentials=True,
    allow_methods=["*"],
    allow_headers=["*"],
)

add_routes(app, chain, path="/chat")

from fastapi import FastAPI, Depends, HTTPException, Security
from fastapi.security import HTTPBearer, HTTPAuthorizationCredentials
from langserve import add_routes

app = FastAPI()
security = HTTPBearer()

def verify_token(credentials: HTTPAuthorizationCredentials = Security(security)):
    """Vérifie le token d'authentification"""
    token = credentials.credentials

    # Vérification (simplifiée)
    if token != "secret-token-123":
        raise HTTPException(status_code=401, detail="Token invalide")

    return token

# Route protégée
add_routes(
    app,
    chain,
    path="/chat",
    dependencies=[Depends(verify_token)]  # Authentification requise
)

from langserve import RemoteRunnable

chat = RemoteRunnable(
    "http://localhost:8000/chat",
    headers={"Authorization": "Bearer secret-token-123"}
)

response = chat.invoke("Question")

from fastapi import FastAPI, Request
from slowapi import Limiter, _rate_limit_exceeded_handler
from slowapi.util import get_remote_address
from slowapi.errors import RateLimitExceeded
from langserve import add_routes

limiter = Limiter(key_func=get_remote_address)
app = FastAPI()
app.state.limiter = limiter
app.add_exception_handler(RateLimitExceeded, _rate_limit_exceeded_handler)

# Option 1 : Rate limiting sur endpoint personnalisé
@app.post("/custom/invoke")
@limiter.limit("10/minute")  # 10 requêtes par minute
async def limited_invoke(request: Request):
    """Endpoint personnalisé avec rate limiting"""
    data = await request.json()

    # Invoke la chain
    result = await chain.ainvoke(data.get("input", {}))

    return {"output": result}

# Option 2 : Utiliser add_routes avec rate limiting global
# via middleware (recommandé pour production)
add_routes(app, chain, path="/chat")

from langserve import add_routes

add_routes(
    app,
    chain,
    path="/chat",
    enabled_endpoints=["invoke", "stream"],  # Désactive batch
    input_type=dict,  # Type d'input
    output_type=str,  # Type d'output
    config_keys=["metadata", "tags"],  # Clés de config autorisées
    enable_feedback_endpoint=True,  # Endpoint de feedback
    enable_public_trace_link_endpoint=False,  # Pas de liens publics
    playground_type="default"  # Type de playground
)

FROM python:3.11-slim

WORKDIR /app

# Dépendances
COPY requirements.txt .
RUN pip install --no-cache-dir -r requirements.txt

# Code
COPY . .

# Port
EXPOSE 8000

# Variables d'environnement
ENV OPENAI_API_KEY=""

# Commande
CMD ["uvicorn", "server:app", "--host", "0.0.0.0", "--port", "8000"]

langchain
langchain-openai
langserve[all]
uvicorn[standard]
python-dotenv

version: '3.8'

services:
  api:
    build: .
    ports:
      - "8000:8000"
    environment:
      - OPENAI_API_KEY=${OPENAI_API_KEY}
      - LANGCHAIN_TRACING_V2=true
      - LANGCHAIN_API_KEY=${LANGCHAIN_API_KEY}
    volumes:
      - ./vectorstore:/app/vectorstore
    restart: unless-stopped

# Build
docker-compose build

# Run
docker-compose up -d

# Logs
docker-compose logs -f

# Stop
docker-compose down

# Installation
npm install -g @railway/cli

# Login
railway login

# Initialisation
railway init

# Déploiement
railway up

{
  "build": {
    "builder": "DOCKERFILE"
  },
  "deploy": {
    "startCommand": "uvicorn server:app --host 0.0.0.0 --port $PORT",
    "restartPolicyType": "ON_FAILURE"
  }
}

services:
  - type: web
    name: langchain-api
    env: python
    buildCommand: pip install -r requirements.txt
    startCommand: uvicorn server:app --host 0.0.0.0 --port $PORT
    envVars:
      - key: OPENAI_API_KEY
        sync: false
      - key: PYTHON_VERSION
        value: 3.11.0

# Build et push
gcloud builds submit --tag gcr.io/PROJECT_ID/langchain-api

# Déploiement
gcloud run deploy langchain-api \
  --image gcr.io/PROJECT_ID/langchain-api \
  --platform managed \
  --region us-central1 \
  --allow-unauthenticated \
  --set-env-vars OPENAI_API_KEY=$OPENAI_API_KEY

# Multiple workers (CPU-bound)
uvicorn server:app --workers 4 --host 0.0.0.0 --port 8000

# Avec Gunicorn
gunicorn server:app -w 4 -k uvicorn.workers.UvicornWorker --bind 0.0.0.0:8000

from fastapi import FastAPI
from langserve import add_routes
from langchain.cache import InMemoryCache
from langchain.globals import set_llm_cache

app = FastAPI()

# Cache global pour LLM
set_llm_cache(InMemoryCache())

add_routes(app, chain, path="/chat")

from fastapi import FastAPI

app = FastAPI()

@app.get("/health")
async def health_check():
    """Endpoint de santé pour load balancers"""
    return {
        "status": "healthy",
        "version": "1.0.0"
    }

add_routes(app, chain, path="/chat")

import logging
from fastapi import FastAPI, Request
import time

# Configuration logging
logging.basicConfig(
    level=logging.INFO,
    format='%(asctime)s - %(name)s - %(levelname)s - %(message)s'
)
logger = logging.getLogger(__name__)

app = FastAPI()

@app.middleware("http")
async def log_requests(request: Request, call_next):
    """Middleware pour logger toutes les requêtes"""
    start_time = time.time()

    response = await call_next(request)

    duration = time.time() - start_time

    logger.info(
        f"{request.method} {request.url.path} "
        f"- Status: {response.status_code} "
        f"- Duration: {duration:.2f}s"
    )

    return response

add_routes(app, chain, path="/chat")

from fastapi import FastAPI
from prometheus_fastapi_instrumentator import Instrumentator

app = FastAPI()

# Instrumenter l'app
Instrumentator().instrument(app).expose(app)

add_routes(app, chain, path="/chat")

# Métriques disponibles sur /metrics

import os
from fastapi import FastAPI

# Configuration LangSmith
os.environ["LANGCHAIN_TRACING_V2"] = "true"
os.environ["LANGCHAIN_API_KEY"] = "ls-..."
os.environ["LANGCHAIN_PROJECT"] = "langserve-production"

app = FastAPI()

# Toutes les requêtes sont automatiquement tracées dans LangSmith
add_routes(app, chain, path="/chat")

from langserve import RemoteRunnable
from typing import Iterator

class LangChainClient:
    """Client wrapper pour LangServe"""

    def __init__(self, base_url: str, api_key: str = None):
        self.base_url = base_url
        self.headers = {}

        if api_key:
            self.headers["Authorization"] = f"Bearer {api_key}"

        self.chat = RemoteRunnable(
            f"{base_url}/chat",
            headers=self.headers
        )

    def ask(self, question: str) -> str:
        """Question simple (invoke)"""
        response = self.chat.invoke(question)
        return response.content

    def ask_stream(self, question: str) -> Iterator[str]:
        """Question avec streaming"""
        for chunk in self.chat.stream(question):
            yield chunk.content

    def ask_batch(self, questions: list[str]) -> list[str]:
        """Batch de questions"""
        responses = self.chat.batch(questions)
        return [r.content for r in responses]

# Utilisation
client = LangChainClient(
    base_url="http://localhost:8000",
    api_key="secret-token"
)

# Simple
answer = client.ask("Qu'est-ce que l'IA ?")
print(answer)

# Streaming
for chunk in client.ask_stream("Raconte une histoire"):
    print(chunk, end="", flush=True)

# Batch
answers = client.ask_batch([
    "Question 1",
    "Question 2",
    "Question 3"
])

import { RemoteRunnable } from "@langchain/core/runnables/remote";

// Configuration
const chat = new RemoteRunnable({
  url: "http://localhost:8000/chat",
  options: {
    headers: {
      "Authorization": "Bearer secret-token"
    }
  }
});

// Invoke
const response = await chat.invoke("Qu'est-ce que l'IA ?");
console.log(response);

// Streaming
const stream = await chat.stream("Raconte une histoire");
for await (const chunk of stream) {
  process.stdout.write(chunk.content);
}

// Batch
const responses = await chat.batch([
  "Question 1",
  "Question 2",
  "Question 3"
]);

from fastapi import FastAPI, Header, HTTPException

app = FastAPI()

# Chains par tenant
chains = {
    "tenant-1": create_chain(config_1),
    "tenant-2": create_chain(config_2),
}

@app.post("/chat/invoke")
async def multi_tenant_invoke(
    request: dict,
    x_tenant_id: str = Header(...)
):
    """Route avec sélection de chain par tenant"""

    if x_tenant_id not in chains:
        raise HTTPException(status_code=404, detail="Tenant non trouvé")

    chain = chains[x_tenant_id]
    result = await chain.ainvoke(request["input"])

    return {"output": result}

from fastapi import FastAPI
from langserve import add_routes
from langchain_openai import ChatOpenAI
from langchain_anthropic import ChatAnthropic

app = FastAPI()

# LLM principal
primary_llm = ChatOpenAI(model="gpt-4", timeout=10)

# LLM de secours
fallback_llm = ChatAnthropic(model="claude-sonnet-4.5")

# Chain avec fallback automatique
chain = primary_llm.with_fallbacks([fallback_llm])

add_routes(app, chain, path="/chat")

from fastapi import FastAPI
from langserve import add_routes

app = FastAPI()

# Version 1 (stable)
chain_v1 = create_chain_v1()
add_routes(app, chain_v1, path="/v1/chat")

# Version 2 (beta)
chain_v2 = create_chain_v2()
add_routes(app, chain_v2, path="/v2/chat")

# Latest (alias vers version stable)
add_routes(app, chain_v1, path="/chat")

from fastapi import FastAPI, HTTPException
from pydantic import BaseModel, Field, validator
from langserve import add_routes

app = FastAPI()

class ChatRequest(BaseModel):
    """Modèle validé pour les requêtes"""
    message: str = Field(..., min_length=1, max_length=2000)

    @validator('message')
    def validate_message(cls, v):
        # Pas de contenu malveillant
        forbidden = ["<script>", "DROP TABLE", "'; DELETE"]
        if any(bad in v for bad in forbidden):
            raise ValueError("Contenu suspect détecté")
        return v

@app.post("/chat/safe")
async def safe_chat(request: ChatRequest):
    """Endpoint avec validation stricte"""
    result = chain.invoke(request.message)
    return {"response": result}

import os
from dotenv import load_dotenv

# Charger depuis .env (dev)
load_dotenv()

# Ou depuis environment variables (prod)
OPENAI_API_KEY = os.getenv("OPENAI_API_KEY")
LANGCHAIN_API_KEY = os.getenv("LANGCHAIN_API_KEY")

# Validation
if not OPENAI_API_KEY:
    raise ValueError("OPENAI_API_KEY manquante")

# Utilisation sécurisée
llm = ChatOpenAI(api_key=OPENAI_API_KEY)

import uvicorn

# Configuration
config = uvicorn.Config(
    app,
    host="0.0.0.0",
    port=8000,
    timeout_keep_alive=300,  # 5 minutes
    timeout_graceful_shutdown=30
)

server = uvicorn.Server(config)
server.run()

from fastapi.middleware.cors import CORSMiddleware

app.add_middleware(
    CORSMiddleware,
    allow_origins=["*"],  # Ou liste spécifique
    allow_credentials=True,
    allow_methods=["*"],
    allow_headers=["*"],
    expose_headers=["*"]
)

import gc
from fastapi import FastAPI

app = FastAPI()

@app.middleware("http")
async def cleanup_middleware(request: Request, call_next):
    response = await call_next(request)

    # Forcer garbage collection périodiquement
    if request.url.path.endswith("/invoke"):
        gc.collect()

    return response

# Installation
pip install langserve[all]

# Lancer serveur
uvicorn server:app --reload

# Avec workers
uvicorn server:app --workers 4

# Docker
docker-compose up -d

# Serveur
from fastapi import FastAPI
from langserve import add_routes
from langchain_openai import ChatOpenAI

app = FastAPI()
add_routes(app, ChatOpenAI(), path="/chat")

# Client
from langserve import RemoteRunnable
chat = RemoteRunnable("http://localhost:8000/chat")
print(chat.invoke("Hello"))

GET  /chat/playground          # Interface interactive
POST /chat/invoke              # Requête simple
POST /chat/batch               # Batch de requêtes
POST /chat/stream              # Streaming SSE
GET  /chat/input_schema        # Schema input
GET  /docs                     # Swagger UI

from langchain_core.prompts import ChatPromptTemplate

prompt = ChatPromptTemplate.from_messages([
    ("system", "Tu es un traducteur expert. Traduis en {target_language}."),
    ("human", "{text}")
])

# TODO: Compléter l'implémentation