integration

LangChain Integration Guide

Integration with vector stores, LangSmith observability, and deployment.

Vector store integrations

Chroma (local, open-source)

from langchain_chroma import Chroma
from langchain_openai import OpenAIEmbeddings

# Create vector store
vectorstore = Chroma.from_documents(
    documents=docs,
    embedding=OpenAIEmbeddings(),
    persist_directory="./chroma_db"
)

# Load existing store
vectorstore = Chroma(
    persist_directory="./chroma_db",
    embedding_function=OpenAIEmbeddings()
)

# Add documents incrementally
vectorstore.add_documents([new_doc1, new_doc2])

# Delete documents
vectorstore.delete(ids=["doc1", "doc2"])

Pinecone (cloud, scalable)

from langchain_pinecone import PineconeVectorStore
import pinecone

# Initialize Pinecone
pinecone.init(api_key="your-api-key", environment="us-west1-gcp")

# Create index (one-time)
pinecone.create_index("my-index", dimension=1536, metric="cosine")

# Create vector store
vectorstore = PineconeVectorStore.from_documents(
    documents=docs,
    embedding=OpenAIEmbeddings(),
    index_name="my-index"
)

# Query with metadata filters
results = vectorstore.similarity_search(
    "Python tutorials",
    k=4,
    filter={"category": "beginner"}
)

FAISS (fast similarity search)

from langchain_community.vectorstores import FAISS

# Create FAISS index
vectorstore = FAISS.from_documents(docs, OpenAIEmbeddings())

# Save to disk
vectorstore.save_local("./faiss_index")

# Load from disk
vectorstore = FAISS.load_local(
    "./faiss_index",
    OpenAIEmbeddings(),
    allow_dangerous_deserialization=True
)

# Merge multiple indices
vectorstore1 = FAISS.load_local("./index1", embeddings)
vectorstore2 = FAISS.load_local("./index2", embeddings)
vectorstore1.merge_from(vectorstore2)

Weaviate (production, ML-native)

from langchain_weaviate import WeaviateVectorStore
import weaviate

# Connect to Weaviate
client = weaviate.Client("http://localhost:8080")

# Create vector store
vectorstore = WeaviateVectorStore.from_documents(
    documents=docs,
    embedding=OpenAIEmbeddings(),
    client=client,
    index_name="LangChain"
)

# Hybrid search (vector + keyword)
results = vectorstore.similarity_search(
    "Python async",
    k=4,
    alpha=0.5  # 0=keyword, 1=vector, 0.5=hybrid
)

Qdrant (fast, open-source)

from langchain_qdrant import QdrantVectorStore
from qdrant_client import QdrantClient

# Connect to Qdrant
client = QdrantClient(host="localhost", port=6333)

# Create vector store
vectorstore = QdrantVectorStore.from_documents(
    documents=docs,
    embedding=OpenAIEmbeddings(),
    collection_name="my_documents",
    client=client
)

LangSmith observability

Enable tracing

import os

# Set environment variables
os.environ["LANGCHAIN_TRACING_V2"] = "true"
os.environ["LANGCHAIN_API_KEY"] = "your-langsmith-api-key"
os.environ["LANGCHAIN_PROJECT"] = "my-project"

# All chains/agents automatically traced
from langchain.agents import create_agent
from langchain_anthropic import ChatAnthropic

agent = create_agent(
    model=ChatAnthropic(model="claude-sonnet-4-5-20250929"),
    tools=[calculator, search]
)

# Run - automatically logged to LangSmith
result = agent.invoke({"input": "What is 25 * 17?"})

# View traces at https://smith.langchain.com

Custom metadata

from langchain.callbacks import tracing_v2_enabled

# Add custom metadata to traces
with tracing_v2_enabled(
    project_name="my-project",
    tags=["production", "customer-support"],
    metadata={"user_id": "12345", "session_id": "abc"}
):
    result = agent.invoke({"input": "Help me with Python"})

Evaluate runs

from langsmith import Client

client = Client()

# Create dataset
dataset = client.create_dataset("qa-eval")
client.create_example(
    dataset_id=dataset.id,
    inputs={"question": "What is Python?"},
    outputs={"answer": "Python is a programming language"}
)

# Evaluate
from langchain.evaluation import load_evaluator

evaluator = load_evaluator("qa")
results = client.evaluate(
    lambda x: qa_chain(x),
    data=dataset,
    evaluators=[evaluator]
)

Deployment patterns

FastAPI server

from fastapi import FastAPI
from pydantic import BaseModel
from langchain.agents import create_agent

app = FastAPI()

# Initialize agent once
agent = create_agent(
    model=llm,
    tools=[search, calculator]
)

class Query(BaseModel):
    input: str

@app.post("/chat")
async def chat(query: Query):
    result = agent.invoke({"input": query.input})
    return {"response": result["output"]}

# Run: uvicorn main:app --reload

Streaming responses

from fastapi.responses import StreamingResponse
from langchain.callbacks import AsyncIteratorCallbackHandler

@app.post("/chat/stream")
async def chat_stream(query: Query):
    callback = AsyncIteratorCallbackHandler()

    async def generate():
        async for token in agent.astream({"input": query.input}):
            if "output" in token:
                yield token["output"]

    return StreamingResponse(generate(), media_type="text/plain")

Docker deployment

# Dockerfile
FROM python:3.11-slim

WORKDIR /app

COPY requirements.txt .
RUN pip install -r requirements.txt

COPY . .

CMD ["uvicorn", "main:app", "--host", "0.0.0.0", "--port", "8000"]

# Build and run
docker build -t langchain-app .
docker run -p 8000:8000 \
  -e OPENAI_API_KEY=your-key \
  -e LANGCHAIN_API_KEY=your-key \
  langchain-app

Kubernetes deployment

# deployment.yaml
apiVersion: apps/v1
kind: Deployment
metadata:
  name: langchain-app
spec:
  replicas: 3
  selector:
    matchLabels:
      app: langchain
  template:
    metadata:
      labels:
        app: langchain
    spec:
      containers:
      - name: langchain
        image: your-registry/langchain-app:latest
        ports:
        - containerPort: 8000