LangChain + DeepSeek完全統合ガイド

LangChain + DeepSeek 概要

LangChain は Python/JavaScript 向けの LLM アプリケーション開発フレームワークです。DeepSeek は OpenAI 互換 API を通じて LangChain のすべてのコンポーネント（Chat Model、Embeddings、Chains、Agents、Memory、Retrievers など）にシームレスに統合できます。base_url を変更するだけで使用できます。

1. 基本設定

# インストール
pip install langchain langchain-openai python-dotenv

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

2. ChatModel 統合

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,
)

# 基本的な会話
messages = [
    SystemMessage(content="あなたは Python の専門家です"),
    HumanMessage(content="デコレータの原理を説明してください")
]
response = llm.invoke(messages)
print(response.content)

3. ストリーミング出力

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,
)

# ストリーミング生成 - トークンごとに出力
llm_stream.invoke("AIについての詩を書いてください")

4. Embeddings 統合

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"),
)

# テキストベクトルの生成
text = "DeepSeek V4 は現在最もコストパフォーマンスの高い AI モデルです"
vector = embeddings.embed_query(text)
print(f"ベクトル次元: {len(vector)}")  # 1536

# バッチ生成
texts = ["AIモデル", "機械学習", "深層学習"]
vectors = embeddings.embed_documents(texts)
print(f"{len(vectors)} 個のベクトルを生成しました")

5. Tool Calling（Agent ツール）

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

@tool
def get_weather(city: str) -> str:
    """都市の天気を取得"""
    # 実際に天気 API を呼び出す
    return f"{city} は今日晴れ、25°C"

@tool
def calculator(expression: str) -> str:
    """数学計算を実行"""
    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("北京の天気はどうですか？ついでに 256*128 を計算してください")
print(result['output'])

6. RAG Chain（検索拡張生成）

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

# 1. ドキュメントの読み込み
loader = TextLoader("knowledge.txt")
docs = loader.load()

# 2. テキストの分割
text_splitter = RecursiveCharacterTextSplitter(chunk_size=500, chunk_overlap=50)
chunks = text_splitter.split_documents(docs)

# 3. ベクトルストアの構築
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. 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. クエリ
result = qa_chain.invoke({"query": "DeepSeek V4 の特徴は何ですか？"})
print(result['result'])
print(f"ソースドキュメント数: {len(result['source_documents'])}")

7. 会話メモリ

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

memory = ConversationSummaryBufferMemory(
    llm=llm,
    max_token_limit=2000,  # 要約圧縮のしきい値
    return_messages=True,
)

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

# マルチターン会話
conversation.predict(input="私の名前は張三です。Python 開発者です")
conversation.predict(input="主に Web バックエンド開発をしています")
conversation.predict(input="私の名前と職業を覚えていますか？")
# 出力: あなたは張三、Python Web バックエンド開発者です

8. 本番環境の最適化

from langchain.callbacks import get_openai_callback

# Token 使用量の監視
with get_openai_callback() as cb:
    response = llm.invoke("RAG とは何か説明してください")
    print(f"Token 消費: {cb.total_tokens}")
    print(f"コスト見積: ¥{cb.total_cost:.4f}")

# キャッシュ（重複呼び出しの削減）
from langchain.cache import InMemoryCache
import langchain

langchain.llm_cache = InMemoryCache()
# 同じプロンプトの2回目の呼び出しはキャッシュから直接返されます