Jakob Wennberg
12 KiB
12 KiB
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
env:
- name: OPENAI_API_KEY
valueFrom:
secretKeyRef:
name: langchain-secrets
key: openai-api-key
resources:
requests:
memory: "512Mi"
cpu: "500m"
limits:
memory: "2Gi"
cpu: "2000m"
Model integrations
OpenAI
from langchain_openai import ChatOpenAI
llm = ChatOpenAI(
model="gpt-4o",
temperature=0,
max_tokens=1000,
timeout=30,
max_retries=2
)
Anthropic
from langchain_anthropic import ChatAnthropic
llm = ChatAnthropic(
model="claude-sonnet-4-5-20250929",
temperature=0,
max_tokens=4096,
timeout=60
)
from langchain_google_genai import ChatGoogleGenerativeAI
llm = ChatGoogleGenerativeAI(
model="gemini-2.0-flash-exp",
temperature=0
)
Local models (Ollama)
from langchain_community.llms import Ollama
llm = Ollama(
model="llama3",
base_url="http://localhost:11434"
)
Azure OpenAI
from langchain_openai import AzureChatOpenAI
llm = AzureChatOpenAI(
azure_endpoint="https://your-endpoint.openai.azure.com/",
azure_deployment="gpt-4",
api_version="2024-02-15-preview"
)
Tool integrations
Web search
from langchain_community.tools import DuckDuckGoSearchRun, TavilySearchResults
# DuckDuckGo (free)
search = DuckDuckGoSearchRun()
# Tavily (best quality)
search = TavilySearchResults(api_key="your-key")
Wikipedia
from langchain_community.tools import WikipediaQueryRun
from langchain_community.utilities import WikipediaAPIWrapper
wikipedia = WikipediaQueryRun(api_wrapper=WikipediaAPIWrapper())
Python REPL
from langchain_experimental.tools import PythonREPLTool
python_repl = PythonREPLTool()
# Agent can execute Python code
agent = create_agent(model=llm, tools=[python_repl])
result = agent.invoke({"input": "Calculate the 10th Fibonacci number"})
Shell commands
from langchain_community.tools import ShellTool
shell = ShellTool()
# Agent can run shell commands
agent = create_agent(model=llm, tools=[shell])
SQL databases
from langchain_community.utilities import SQLDatabase
from langchain_community.agent_toolkits import create_sql_agent
db = SQLDatabase.from_uri("sqlite:///mydatabase.db")
agent = create_sql_agent(
llm=llm,
db=db,
agent_type="openai-tools",
verbose=True
)
result = agent.run("How many users are in the database?")
Memory integrations
Redis
from langchain.memory import RedisChatMessageHistory
from langchain.memory import ConversationBufferMemory
# Redis-backed memory
message_history = RedisChatMessageHistory(
url="redis://localhost:6379",
session_id="user-123"
)
memory = ConversationBufferMemory(
chat_memory=message_history,
return_messages=True
)
PostgreSQL
from langchain_postgres import PostgresChatMessageHistory
message_history = PostgresChatMessageHistory(
connection_string="postgresql://user:pass@localhost/db",
session_id="user-123"
)
MongoDB
from langchain_mongodb import MongoDBChatMessageHistory
message_history = MongoDBChatMessageHistory(
connection_string="mongodb://localhost:27017/",
session_id="user-123"
)
Caching
In-memory cache
from langchain.cache import InMemoryCache
from langchain.globals import set_llm_cache
set_llm_cache(InMemoryCache())
# Same query uses cache
response1 = llm.invoke("What is Python?") # API call
response2 = llm.invoke("What is Python?") # Cached
SQLite cache
from langchain.cache import SQLiteCache
set_llm_cache(SQLiteCache(database_path=".langchain.db"))
Redis cache
from langchain.cache import RedisCache
from redis import Redis
set_llm_cache(RedisCache(redis_=Redis(host="localhost", port=6379)))
Monitoring & logging
Custom callbacks
from langchain.callbacks.base import BaseCallbackHandler
class CustomCallback(BaseCallbackHandler):
def on_llm_start(self, serialized, prompts, **kwargs):
print(f"LLM started with prompts: {prompts}")
def on_llm_end(self, response, **kwargs):
print(f"LLM finished with: {response}")
def on_tool_start(self, serialized, input_str, **kwargs):
print(f"Tool {serialized['name']} started with: {input_str}")
def on_tool_end(self, output, **kwargs):
print(f"Tool finished with: {output}")
# Use callback
agent = create_agent(
model=llm,
tools=[calculator],
callbacks=[CustomCallback()]
)
Token counting
from langchain.callbacks import get_openai_callback
with get_openai_callback() as cb:
result = llm.invoke("Write a long story")
print(f"Tokens used: {cb.total_tokens}")
print(f"Cost: ${cb.total_cost:.4f}")
Best practices
- Use LangSmith in production - Essential for debugging
- Cache aggressively - LLM calls are expensive
- Set timeouts - Prevent hanging requests
- Add retries - Handle transient failures
- Monitor costs - Track token usage
- Version your prompts - Track changes
- Use async - Better performance for I/O
- Persistent memory - Don't lose conversation history
- Secure API keys - Use environment variables
- Test integrations - Verify connections before production
Resources
- LangSmith: https://smith.langchain.com
- Vector Stores: https://python.langchain.com/docs/integrations/vectorstores
- Model Providers: https://python.langchain.com/docs/integrations/llms
- Tools: https://python.langchain.com/docs/integrations/tools
- Deployment Guide: https://docs.langchain.com/deploy