LangChain + DeepSeek Overview

LangChain is an LLM application development framework for Python/JavaScript. DeepSeek can seamlessly integrate with all LangChain components via an OpenAI-compatible API: Chat Model, Embeddings, Chains, Agents, Memory, Retrievers, etc. You only need to modify the base_url to use it.

1. Basic Configuration

# Installation
pip install langchain langchain-openai python-dotenv

# Configure .env
DEEPSEEK_API_KEY=sk-your-key
DEEPSEEK_BASE_URL=https://api.deepseek.com

2. ChatModel Integration

from langchain_openai import ChatOpenAI
from langchain.schema import SystemMessage, HumanMessage
import os
from dotenv import load_dotenv

load_dotenv()

llm = ChatOpenAI(
    model="deepseek-v4-flash",
    openai_api_key=os.getenv("DEEPSEEK_API_KEY"),
    openai_api_base=os.getenv("DEEPSEEK_BASE_URL"),
    temperature=0.7,
    max_tokens=2048,
)

# Basic conversation
messages = [
    SystemMessage(content="You are a Python expert"),
    HumanMessage(content="Explain the principle of decorators")
]
response = llm.invoke(messages)
print(response.content)

3. Streaming Output

from langchain.callbacks.streaming_stdout import StreamingStdOutCallbackHandler

llm_stream = ChatOpenAI(
    model="deepseek-v4-flash",
    openai_api_key=os.getenv("DEEPSEEK_API_KEY"),
    openai_api_base=os.getenv("DEEPSEEK_BASE_URL"),
    streaming=True,
    callbacks=[StreamingStdOutCallbackHandler()],
    temperature=0.7,
)

# Streaming generation - token by token output
llm_stream.invoke("Write a poem about AI")

4. Embeddings Integration

from langchain_openai import OpenAIEmbeddings

embeddings = OpenAIEmbeddings(
    model="text-embedding-3-small",
    openai_api_key=os.getenv("DEEPSEEK_API_KEY"),
    openai_api_base=os.getenv("DEEPSEEK_BASE_URL"),
)

# Generate text vector
text = "DeepSeek V4 is currently the most cost-effective AI model"
vector = embeddings.embed_query(text)
print(f"Vector dimension: {len(vector)}")  # 1536

# Batch generation
texts = ["AI models", "Machine learning", "Deep learning"]
vectors = embeddings.embed_documents(texts)
print(f"Generated {len(vectors)} vectors")

5. Tool Calling (Agent Tools)

from langchain.tools import tool
from langchain.agents import initialize_agent, AgentType

@tool
def get_weather(city: str) -> str:
    """Get city weather"""
    # Actually call weather API
    return f"{city} is sunny today, 25°C"

@tool
def calculator(expression: str) -> str:
    """Perform mathematical calculation"""
    return str(eval(expression))

tools = [get_weather, calculator]

agent = initialize_agent(
    tools=tools,
    llm=llm,
    agent=AgentType.ZERO_SHOT_REACT_DESCRIPTION,
    verbose=True,
    handle_parsing_errors=True,
)

result = agent.invoke("What's the weather in Beijing? Also calculate 256*128")
print(result['output'])

6. RAG Chain (Retrieval-Augmented Generation)

from langchain.document_loaders import TextLoader
from langchain.text_splitter import RecursiveCharacterTextSplitter
from langchain.vectorstores import Chroma
from langchain.chains import RetrievalQA

# 1. Load documents
loader = TextLoader("knowledge.txt")
docs = loader.load()

# 2. Split text
text_splitter = RecursiveCharacterTextSplitter(chunk_size=500, chunk_overlap=50)
chunks = text_splitter.split_documents(docs)

# 3. Build vector store
vectorstore = Chroma.from_documents(
    documents=chunks,
    embedding=OpenAIEmbeddings(
        model="text-embedding-3-small",
        openai_api_key=os.getenv("DEEPSEEK_API_KEY"),
        openai_api_base=os.getenv("DEEPSEEK_BASE_URL"),
    ),
    persist_directory="./chroma_db"
)

# 4. Create RAG Chain
qa_chain = RetrievalQA.from_chain_type(
    llm=llm,
    chain_type="stuff",
    retriever=vectorstore.as_retriever(search_kwargs={"k": 4}),
    return_source_documents=True,
)

# 5. Query
result = qa_chain.invoke({"query": "What are the features of DeepSeek V4?"})
print(result['result'])
print(f"Number of source documents: {len(result['source_documents'])}")

7. Conversation Memory

from langchain.memory import ConversationSummaryBufferMemory
from langchain.chains import ConversationChain

memory = ConversationSummaryBufferMemory(
    llm=llm,
    max_token_limit=2000,  # Summary compression threshold
    return_messages=True,
)

conversation = ConversationChain(
    llm=llm,
    memory=memory,
    verbose=True,
)

# Multi-turn conversation
conversation.predict(input="My name is Zhang San, I am a Python developer")
conversation.predict(input="I mainly do Web backend development")
conversation.predict(input="Do you remember my name and occupation?")
# Output: You are Zhang San, a Python Web backend developer

8. Production Environment Optimization

from langchain.callbacks import get_openai_callback

# Token usage monitoring
with get_openai_callback() as cb:
    response = llm.invoke("Explain what RAG is")
    print(f"Token consumption: {cb.total_tokens}")
    print(f"Cost estimate: ¥{cb.total_cost:.4f}")

# Caching (reduce duplicate calls)
from langchain.cache import InMemoryCache
import langchain

langchain.llm_cache = InMemoryCache()
# The second call with the same prompt will return directly from cache