Gemini × LangChain 統合ガイド — RAG・チェーン・エージェントの構築

LangChain は Gemini を強力に活用するための最適なフレームワークです。このガイドでは、基本的なセットアップからエージェント構築まで、実践的なステップを解説します。

環境セットアップと基本設定

必要なパッケージのインストール

pip install langchain langchain-google-genai chromadb python-dotenv

API キーの設定

import os
from dotenv import load_dotenv
 
load_dotenv()
google_api_key = os.getenv("GOOGLE_API_KEY")
os.environ["GOOGLE_API_KEY"] = google_api_key

ℹ️

Google API キーは [Google AI Studio](https://aistudio.google.com/app/apikey) から取得できます。プロダクション環境では環境変数で安全に管理してください。

基本的な会話モデルの設定

ChatGoogleGenerativeAI の初期化

from langchain_google_genai import ChatGoogleGenerativeAI
 
# 基本的なモデル初期化
model = ChatGoogleGenerativeAI(
    model="gemini-2.5-pro",
    temperature=0.7,
    max_tokens=1024
)
 
# シンプルな会話
response = model.invoke("Python で簡単な計算器の実装方法を教えてください")
print(response.content)

ストリーミング出力

# ストリーミング対応の初期化
model_streaming = ChatGoogleGenerativeAI(
    model="gemini-2.5-flash",
    streaming=True,
    temperature=0.5
)
 
# トークンごとに出力
for chunk in model_streaming.stream("機械学習の基礎を5分で解説してください"):
    print(chunk.content, end="", flush=True)

⚠️

ストリーミングを有効にする場合、API レスポンスが段階的に返却されるため、全体の処理時間が短縮されます。ただしエラーハンドリングが複雑になるため、用途に応じて使い分けてください。

プロンプトテンプレートとチェーン構築

基本的なプロンプトテンプレート

from langchain.prompts import ChatPromptTemplate
 
# テンプレート定義
template = """あなたは {role} です。
ユーザーのお悩みに対して、親切かつ専門的な回答を提供してください。
 
質問: {question}
"""
 
prompt = ChatPromptTemplate.from_template(template)
 
# チェーン構築（LangChain Expression Language）
chain = prompt | model
 
# 実行
result = chain.invoke({
    "role": "Python エキスパート",
    "question": "非同期プログラミングの基本は何ですか？"
})
print(result.content)

複数ステップのチェーン

from langchain.prompts import ChatPromptTemplate
from langchain.output_parsers import StrOutputParser
 
# ステップ1: 要件分析
analysis_template = """以下の要件を分析し、実装のキーポイントを列挙してください。
 
要件:
{requirement}
 
分析結果:"""
 
analysis_prompt = ChatPromptTemplate.from_template(analysis_template)
analysis_chain = analysis_prompt | model | StrOutputParser()
 
# ステップ2: 実装案の生成
impl_template = """以下の分析に基づいて、Python による実装方法を提案してください。
 
分析結果:
{analysis}
 
実装方法:"""
 
impl_prompt = ChatPromptTemplate.from_template(impl_template)
impl_chain = impl_prompt | model | StrOutputParser()
 
# パイプライン構築
full_chain = {
    "requirement": lambda x: x,
    "analysis": analysis_chain
} | (lambda x: impl_chain.invoke(x))
 
# 実行
requirement = "キャッシング機能付きの API クライアントを実装したい"
result = full_chain(requirement)
print(result)

RAG パイプラインの構築

データストアの準備

from langchain.document_loaders import TextLoader
from langchain.text_splitter import RecursiveCharacterTextSplitter
from langchain_google_genai import GoogleGenerativeAIEmbeddings
from langchain.vectorstores import Chroma
 
# ドキュメント読み込み
loader = TextLoader("documents/gemini_guide.txt")
documents = loader.load()
 
# テキスト分割
splitter = RecursiveCharacterTextSplitter(
    chunk_size=1000,
    chunk_overlap=200
)
splits = splitter.split_documents(documents)
 
# Embedding モデルの初期化
embeddings = GoogleGenerativeAIEmbeddings(
    model="models/embedding-001"
)
 
# ベクトルデータベースの作成
vectorstore = Chroma.from_documents(
    documents=splits,
    embedding=embeddings,
    persist_directory="./chroma_db"
)

RAG チェーンの構築

from langchain.chains import create_retrieval_chain
from langchain.chains.combine_documents import create_stuff_documents_chain
from langchain.prompts import ChatPromptTemplate
 
# Retriever の設定
retriever = vectorstore.as_retriever(
    search_type="similarity",
    search_kwargs={"k": 3}
)
 
# RAG 用プロンプト
rag_prompt = ChatPromptTemplate.from_template("""以下のドキュメントに基づいて質問に回答してください。
 
<context>
{context}
</context>
 
質問: {input}
 
回答:""")
 
# RAG チェーン構築
combine_docs_chain = create_stuff_documents_chain(model, rag_prompt)
rag_chain = create_retrieval_chain(retriever, combine_docs_chain)
 
# クエリ実行
result = rag_chain.invoke({
    "input": "Gemini の高度な機能は何ですか？"
})
print(result["answer"])

ℹ️

ChromaDB は開発環境に最適なメモリベースのベクトルストアです。プロダクション環境では Vertex AI Vector Search や Weaviate 等の専用サービスの使用を推奨します。

ReAct エージェントの構築

ツール定義

from langchain.tools import tool
from typing import Any
 
@tool
def search_web(query: str) -> str:
    """ウェブで検索する"""
    # 実装例（実際には API を使用）
    return f"'{query}' に関する検索結果..."
 
@tool
def calculate(expression: str) -> float:
    """数式を計算する"""
    try:
        return eval(expression)
    except Exception as e:
        return f"計算エラー: {str(e)}"
 
@tool
def get_current_time() -> str:
    """現在の時刻を取得する"""
    from datetime import datetime
    return datetime.now().isoformat()
 
tools = [search_web, calculate, get_current_time]

エージェントの構築と実行

from langchain.agents import create_tool_using_agent_executor
from langchain.agents import AgentType
from langchain.agents import initialize_agent
 
# ReAct エージェントの初期化
agent = initialize_agent(
    tools=tools,
    llm=model,
    agent=AgentType.ZERO_SHOT_REACT_DESCRIPTION,
    verbose=True,
    max_iterations=10,
    handle_parsing_errors=True
)
 
# エージェントの実行
response = agent.run(
    "現在の時刻から1時間後は何時ですか？ また、その時刻の天気予報を検索してください。"
)
print(response)

カスタムツールによる複雑なワークフロー

from langchain.tools import tool
from pydantic import BaseModel, Field
 
class SearchInput(BaseModel):
    query: str = Field(description="検索キーワード")
    filters: str = Field(default="", description="検索フィルタ")
 
@tool(args_schema=SearchInput)
def advanced_search(query: str, filters: str = "") -> str:
    """高度な検索機能"""
    result = f"クエリ: {query}"
    if filters:
        result += f" (フィルタ: {filters})"
    return result
 
# エージェント用ツール更新
custom_tools = [advanced_search, calculate, get_current_time]
 
agent = initialize_agent(
    tools=custom_tools,
    llm=model,
    agent=AgentType.ZERO_SHOT_REACT_DESCRIPTION,
    verbose=True
)

⚠️

エージェントは無限ループのリスクがあります。`max_iterations` と `max_execution_time` を必ず設定し、予期しない動作を防いでください。

メモリ管理とコンテキスト保持

会話履歴の管理

from langchain.memory import ConversationBufferMemory
from langchain.chains import ConversationChain
 
# メモリの初期化
memory = ConversationBufferMemory()
 
# 会話チェーン構築
conversation = ConversationChain(
    llm=model,
    memory=memory,
    verbose=True
)
 
# 複数の会話を実行
response1 = conversation.predict(input="私の名前は田中です。")
response2 = conversation.predict(input="私の名前は何ですか？")
response3 = conversation.predict(input="Pythonについて教えてください。")
 
print(memory.buffer)

トークン数制限付きメモリ

from langchain.memory import ConversationTokenBufferMemory
 
# トークン数制限メモリ
memory = ConversationTokenBufferMemory(
    llm=model,
    max_token_limit=1000
)
 
conversation = ConversationChain(
    llm=model,
    memory=memory,
    verbose=True
)
 
# 使用方法は同じ
response = conversation.predict(input="継続的な会話をサポートします。")

エラーハンドリング

例外処理のベストプラクティス

from langchain_google_genai import ChatGoogleGenerativeAI
import logging
 
logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)
 
def safe_invoke(model: ChatGoogleGenerativeAI, prompt: str) -> str:
    """安全なモデル呼び出し"""
    try:
        response = model.invoke(prompt)
        return response.content
    except Exception as e:
        logger.error(f"API エラー: {str(e)}")
        return "申し訳ありません。エラーが発生しました。"
 
# 使用例
result = safe_invoke(model, "これはテスト質問です")
print(result)

リトライロジック

from tenacity import retry, stop_after_attempt, wait_exponential
 
@retry(
    stop=stop_after_attempt(3),
    wait=wait_exponential(multiplier=1, min=2, max=10)
)
def invoke_with_retry(model, prompt):
    """リトライ付きの呼び出し"""
    return model.invoke(prompt).content
 
try:
    result = invoke_with_retry(model, "リトライ可能なクエリ")
except Exception as e:
    logger.error(f"最大リトライ回数に達しました: {str(e)}")

パフォーマンス最適化

バッチ処理

from typing import List
 
def batch_invoke(model: ChatGoogleGenerativeAI, prompts: List[str]) -> List[str]:
    """複数のプロンプトをバッチ処理"""
    results = []
    for prompt in prompts:
        try:
            response = model.invoke(prompt)
            results.append(response.content)
        except Exception as e:
            logger.warning(f"プロンプト処理失敗: {prompt[:50]}...")
            results.append("")
    return results
 
# 使用例
prompts = [
    "Python の型ヒントについて説明してください",
    "非同期プログラミングのメリットを5つ挙げてください",
    "FastAPI でのエラーハンドリング方法"
]
 
results = batch_invoke(model, prompts)
for prompt, result in zip(prompts, results):
    print(f"Q: {prompt[:30]}...\nA: {result[:100]}...\n")

非同期処理

import asyncio
from langchain_google_genai import ChatGoogleGenerativeAI
 
async def async_invoke(model: ChatGoogleGenerativeAI, prompts: List[str]) -> List[str]:
    """非同期複数処理"""
    tasks = [
        asyncio.to_thread(model.invoke, prompt)
        for prompt in prompts
    ]
    responses = await asyncio.gather(*tasks)
    return [r.content for r in responses]
 
# 実行
if __name__ == "__main__":
    prompts = [
        "機械学習の基礎",
        "データ分析のベストプラクティス",
        "API 設計のアンチパターン"
    ]
    results = asyncio.run(async_invoke(model, prompts))

全体を振り返って

LangChain と Gemini の組み合わせは、以下のような強力なアプリケーション構築を実現します：

RAG パイプライン: 社内知識ベースを活用したスマート Q&A システム
エージェント: 複数のツールを自動的に活用する自律的なエージェント
チェーン: 複雑な処理フローの構築と管理
メモリ管理: 長期にわたる会話コンテキストの保持

実装時は API レート制限、エラーハンドリング、トークン管理を常に考慮しましょう。

個人開発の視点で — LangChain は「薄く」使う

個人開発で Dolice Labs のツール群を回している私自身、LangChain を使い始めて学んだのは、便利な抽象に乗りすぎると、いざ不具合が出たときに中で何が起きているか追えなくなる、ということでした。RAG もエージェントも、最初は最小構成で自分の手で組み、必要な部分だけ LangChain に任せるほうが結局は安定します。

私はチェーンの各ステップで必ず中間出力をログに残すようにしています。どのリトリーバが何を返し、どのプロンプトで崩れたか——それが見えるだけで、問題の切り分け時間が何分の一にもなります。

RAG の精度が出ないときも、原因の多くはモデルではなく検索側にあります。私はまず取得できた文脈そのものを目視で確認し、欲しい情報が候補に入っているかを見ます。入っていなければチャンク分割や埋め込みを、入っているのに使われなければプロンプトを——と、切り分けてから手を入れるようにしています。

Gemini × LangChain 統合ガイド — RAG・チェーン・エージェントの構築

Gemini × LangChain 統合ガイド — RAG・チェーン・エージェントの構築

環境セットアップと基本設定

必要なパッケージのインストール

API キーの設定

基本的な会話モデルの設定

ChatGoogleGenerativeAI の初期化

ストリーミング出力

プロンプトテンプレートとチェーン構築

基本的なプロンプトテンプレート

複数ステップのチェーン

RAG パイプラインの構築

データストアの準備

RAG チェーンの構築

ReAct エージェントの構築

ツール定義

エージェントの構築と実行

カスタムツールによる複雑なワークフロー

メモリ管理とコンテキスト保持

会話履歴の管理

トークン数制限付きメモリ

エラーハンドリング

例外処理のベストプラクティス

リトライロジック

パフォーマンス最適化

バッチ処理

非同期処理

全体を振り返って

個人開発の視点で — LangChain は「薄く」使う

お読みいただきありがとうございます

関連記事