LangServe

# 1) Install
# pip install 'langserve[all]' langchain langchain-openai langchain-core fastapi uvicorn

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
from langchain_core.runnables import RunnablePassthrough

# 2) Build any LangChain runnable
llm = ChatOpenAI(model='gpt-4o-mini')
prompt = ChatPromptTemplate.from_template('Translate the following to {language}: {text}')
parser = StrOutputParser()

chain = prompt | llm | parser

# 3) Mount it
app = FastAPI(title='Translation API', version='1.0.0')
add_routes(
    app,
    chain,
    path='/translate',
)

# Run: uvicorn main:app --host 0.0.0.0 --port 8000

# 4) What you get for free
# POST /translate/invoke              → single call
# POST /translate/batch                → multiple inputs
# POST /translate/stream               → server-sent events stream
# POST /translate/stream_log           → intermediate steps + tokens
# GET  /translate/input_schema         → JSON Schema for input
# GET  /translate/output_schema        → JSON Schema for output
# GET  /translate/playground/          → interactive web UI
# GET  /docs                            → Swagger / OpenAPI

# 5) Calling from a client
import requests

requests.post('http://localhost:8000/translate/invoke', json={
    'input': { 'language': 'French', 'text': 'Hello, world' },
}).json()
# { 'output': 'Bonjour, le monde', 'metadata': {...} }

# Batch
requests.post('http://localhost:8000/translate/batch', json={
    'inputs': [
        { 'language': 'French',  'text': 'Hello' },
        { 'language': 'Spanish', 'text': 'Hello' },
    ],
}).json()

# 6) Streaming from a client
import requests
with requests.post('http://localhost:8000/translate/stream', json={
    'input': { 'language': 'German', 'text': 'Tell me a long story' },
}, stream=True) as r:
    for line in r.iter_lines():
        if line and line.startswith(b'data: '):
            print(line[6:].decode())

# Or use the LangServe client
from langserve import RemoteRunnable
remote = RemoteRunnable('http://localhost:8000/translate')
for chunk in remote.stream({ 'language': 'German', 'text': 'Long story' }):
    print(chunk, end='', flush=True)

# 7) Auth + middleware (FastAPI native)
from fastapi import Depends, HTTPException, Header

async def verify_api_key(authorization: str = Header(None)):
    if not authorization or not authorization.startswith('Bearer '):
        raise HTTPException(status_code=401)
    if authorization[7:] != os.environ['API_KEY']:
        raise HTTPException(status_code=403)

app = FastAPI(dependencies=[Depends(verify_api_key)])
add_routes(app, chain, path='/translate')

# 8) Per-route configuration
add_routes(
    app,
    chain,
    path='/translate',
    config_keys=['configurable'],         # expose Runnable configurables
    playground_type='default',
    enabled_endpoints=['invoke', 'stream'],   # hide batch + playground
)

# 9) Configurable runnables — choose model at runtime
from langchain_core.runnables import ConfigurableField
llm_configurable = ChatOpenAI(model='gpt-4o-mini').configurable_alternatives(
    ConfigurableField(id='llm'),
    default_key='small',
    big=ChatOpenAI(model='gpt-4o'),
)
chain = prompt | llm_configurable | parser

# Client picks:
requests.post('/translate/invoke', json={
    'input':  { 'language': 'French', 'text': 'Hi' },
    'config': { 'configurable': { 'llm': 'big' } },
})

# 10) Multiple chains in one app
add_routes(app, summarise_chain,  path='/summarise')
add_routes(app, qa_chain,         path='/qa')
add_routes(app, recommend_chain,  path='/recommend')

# 11) Production checklist
# • Lock OpenAPI to the schemas LangServe emits
# • Rate limiting via slowapi or a reverse proxy
# • Caching via langchain.globals.set_llm_cache + Redis backend
# • Streaming: ensure your reverse proxy (nginx, Cloud Run) doesn't buffer SSE
# • Async LangChain models for high concurrency (FastAPI is async)
# • Errors mapped to HTTP codes (LangServe does sensible defaults; override for custom)

# 12) Tracing + observability
# Set LANGCHAIN_TRACING_V2=true + LANGCHAIN_API_KEY for LangSmith
import os
os.environ['LANGCHAIN_TRACING_V2'] = 'true'
os.environ['LANGCHAIN_PROJECT']     = 'translate-prod'
# Every request appears in LangSmith with traces, latency, cost.

# 13) Streaming over WebSockets (alternative to SSE)
# Not built-in; wrap with FastAPI WebSocket and call chain.astream / chain.astream_log yourself.
from fastapi import WebSocket
@app.websocket('/ws/translate')
async def ws(socket: WebSocket):
    await socket.accept()
    while True:
        data = await socket.receive_json()
        async for chunk in chain.astream(data):
            await socket.send_text(chunk)

# 14) Deploying
# • Dockerfile: FROM python:3.12-slim; pip install; CMD uvicorn main:app
# • Cloud Run / Cloud Functions: stateless; scale to zero
# • Kubernetes: HPA on QPS or CPU; readiness probe on '/docs'
# • Fly.io / Render / Railway: drop-in Python service
# • Vercel functions for low-volume; bigger workloads need real containers

# 15) Schema validation
# LangServe auto-generates Pydantic schemas from your prompt template variables + Runnable types.
# For custom input shape, declare types on the chain:
from langchain_core.runnables import Runnable
from pydantic import BaseModel

class TranslateRequest(BaseModel):
    language: str
    text:     str

chain_typed: Runnable[TranslateRequest, str] = chain
add_routes(app, chain_typed, path='/translate-typed')

# 16) Local development
# • langchain serve  (CLI) — quick scaffolding
# • Hot reload: uvicorn main:app --reload
# • Playground at /<path>/playground for ad-hoc testing

# 17) Common bugs
# • Streaming endpoint buffered by proxy → no chunks until response ends; disable buffering in nginx/cloud run
# • async Runnable returns coroutine; FastAPI's sync routes block — call .ainvoke / .astream
# • Mounting two chains at the same path → silent override; pick unique paths
# • OpenAPI playground hidden behind auth → enable per-environment
# • Missing LANGCHAIN_TRACING_V2 → no traces; check the env var spelled correctly
// • Cold start dominates latency → keep min instances or use a warm pool
# • Schema export breaks on union types — pin Pydantic v2 + LangChain compatible versions
# • Multiple workers (gunicorn) without sticky sessions — state cached per worker; design stateless or share via Redis

# main.py
from fastapi import FastAPI
from langserve import add_routes
app = FastAPI()
add_routes(app, chain, path='/chat')
# uvicorn main:app