langchain-architecture

サポート資料の入手可能性

このカタログのコピーには、以下のインラインガイダンスが含まれていますが、次のサポート用パスは含まれていません。これらは同梱ファイルではなく、外部の前提条件です。

• <support-root>/references/agents.md
• <support-root>/references/memory.md
• <support-root>/references/chains.md
• <support-root>/references/document-processing.md
• <support-root>/references/callbacks.md
• <support-root>/assets/agent-template.py
• <support-root>/assets/memory-config.yaml
• <support-root>/assets/chain-example.py

これらのパスに依存する手順を実行する前に、元のサポート資料または既存のインストール環境を入手してください。このカタログには、確認済みのダウンロード元は記載されていません。各パスをそのインストール環境を基準に解決し、内容を確認したうえで、該当するライセンスと実行要件を確認してください。必要なファイルが存在することを確認するまでは、コマンドを実行したり、参照されている手順を実行したと主張したりしないでください。不足しているスクリプトを捏造したり、同じ名前の無関係なファイルを同等のものとして扱ったりしないでください。ここに完全に記載されている独立した手順は、ユーザーが許可した範囲内で実行できます。依存する手順は利用できないものとして報告してください。

LangChain アーキテクチャ

エージェント、チェーン、メモリ、ツール連携を備えた高度な LLM アプリケーションを構築するために、LangChain フレームワークを習得しましょう。

このスキルを使用しない場合

• タスクが LangChain アーキテクチャに関係しない場合
• このスコープ外の別のドメインやツールが必要な場合

手順

• 目標、制約、必要な入力を明確にします。
• 関連するベストプラクティスを適用し、結果を検証します。
• 実行可能な手順と検証方法を提示します。

このスキルを使用する場合

• ツールにアクセスできる自律型 AI エージェントを構築する場合
• 複雑な複数ステップの LLM ワークフローを実装する場合
• 会話のメモリと状態を管理する場合
• LLM を外部データソースや API と連携する場合
• モジュール式で再利用可能な LLM アプリケーションのコンポーネントを作成する場合
• ドキュメント処理パイプラインを実装する場合
• 本番環境向けの LLM アプリケーションを構築する場合

コアコンセプト

1. エージェント

LLM を使用して実行するアクションを決定する自律型システムです。

エージェントの種類：

• ReAct：推論と実行を交互に行います
• OpenAI Functions：関数呼び出し API を活用します
• Structured Chat：複数入力のツールに対応します
• Conversational：チャットインターフェース向けに最適化されています
• Self-Ask with Search：複雑なクエリを分解します

2. チェーン

LLM またはその他のユーティリティを呼び出す一連の処理です。

チェーンの種類：

• LLMChain：基本的なプロンプトと LLM の組み合わせ
• SequentialChain：複数のチェーンを順番に実行します
• RouterChain：入力を専門のチェーンに振り分けます
• TransformChain：各ステップ間でデータを変換します
• MapReduceChain：集約を伴う並列処理を行います

3. メモリ

やり取りをまたいでコンテキストを維持するシステムです。

メモリの種類：

• ConversationBufferMemory：すべてのメッセージを保存します
• ConversationSummaryMemory：古いメッセージを要約します
• ConversationBufferWindowMemory：直近の N 件のメッセージを保持します
• EntityMemory：エンティティに関する情報を追跡します
• VectorStoreMemory：意味的類似性に基づいて検索します

4. ドキュメント処理

検索に使用するドキュメントの読み込み、変換、保存を行います。

コンポーネント：

• Document Loaders：さまざまなソースから読み込みます
• Text Splitters：ドキュメントを適切に分割します
• Vector Stores：埋め込みを保存および取得します
• Retrievers：関連するドキュメントを取得します
• Indexes：効率的にアクセスできるようドキュメントを整理します

5. コールバック

ロギング、監視、デバッグのためのフックです。

ユースケース：

• リクエストとレスポンスのロギング
• トークン使用量の追跡
• レイテンシの監視
• エラー処理
• カスタムメトリクスの収集

バージョンと実行範囲

インラインコードでは、レガシーな LangChain API／インポートを使用しています。LangChain v1 では、レガシーなチェーンと関連機能が langchain-classic に移動されています。最新版をインストールして、これらのスニペットが変更なしで実行できると思い込まないでください。新しいコードでは、公式の v1 移行ガイドとサポート対象のエージェント／状態 API を使用してください：https://docs.langchain.com/oss/python/migrate/langchain-v1 。互換性のあるプロバイダーパッケージを固定し、インポートを確認してください。以下のツールはプレースホルダーであり、実際のデータベースやメール連携ではありません。検証済みの認可された副作用がない限り、メールを送信したと報告しないでください。反復回数、トークン数、有料ツールの使用を制限してください。テナントごとのメモリを分離し、取得したテキストは信頼できないものとして扱ってください。ログではプロンプト、結果、認証情報を必ずマスキングしてください。

クイックスタート

from langchain.agents import AgentType, initialize_agent, load_tools
from langchain.llms import OpenAI
from langchain.memory import ConversationBufferMemory

# Initialize LLM
llm = OpenAI(temperature=0)

# Load tools
tools = load_tools(["serpapi", "llm-math"], llm=llm)

# Add memory
memory = ConversationBufferMemory(memory_key="chat_history")

# Create agent
agent = initialize_agent(
    tools,
    llm,
    agent=AgentType.CONVERSATIONAL_REACT_DESCRIPTION,
    memory=memory,
    verbose=True
)

# Run agent
result = agent.run("What's the weather in SF? Then calculate 25 * 4")

アーキテクチャパターン

パターン 1：LangChain を使用した RAG

from langchain.chains import RetrievalQA
from langchain.document_loaders import TextLoader
from langchain.text_splitter import CharacterTextSplitter
from langchain.vectorstores import Chroma
from langchain.embeddings import OpenAIEmbeddings

# Load and process documents
loader = TextLoader('documents.txt')
documents = loader.load()

text_splitter = CharacterTextSplitter(chunk_size=1000, chunk_overlap=200)
texts = text_splitter.split_documents(documents)

# Create vector store
embeddings = OpenAIEmbeddings()
vectorstore = Chroma.from_documents(texts, embeddings)

# Create retrieval chain
qa_chain = RetrievalQA.from_chain_type(
    llm=llm,
    chain_type="stuff",
    retriever=vectorstore.as_retriever(),
    return_source_documents=True
)

# Query
result = qa_chain({"query": "What is the main topic?"})

パターン 2：ツールを使用したカスタムエージェント

from langchain.agents import Tool, AgentExecutor
from langchain.agents.react.base import ReActDocstoreAgent
from langchain.tools import tool

@tool
def search_database(query: str) -> str:
    """Search internal database for information."""
    # Your database search logic
    return f"Results for: {query}"

@tool
def send_email(recipient: str, content: str) -> str:
    """Send an email to specified recipient."""
    # Email sending logic
    raise NotImplementedError("Configure an authorized email adapter and verify delivery before reporting success")

tools = [search_database, send_email]

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

パターン 3：複数ステップのチェーン

from langchain.chains import LLMChain, SequentialChain
from langchain.prompts import PromptTemplate

# Step 1: Extract key information
extract_prompt = PromptTemplate(
    input_variables=["text"],
    template="Extract key entities from: {text}\n\nEntities:"
)
extract_chain = LLMChain(llm=llm, prompt=extract_prompt, output_key="entities")

# Step 2: Analyze entities
analyze_prompt = PromptTemplate(
    input_variables=["entities"],
    template="Analyze these entities: {entities}\n\nAnalysis:"
)
analyze_chain = LLMChain(llm=llm, prompt=analyze_prompt, output_key="analysis")

# Step 3: Generate summary
summary_prompt = PromptTemplate(
    input_variables=["entities", "analysis"],
    template="Summarize:\nEntities: {entities}\nAnalysis: {analysis}\n\nSummary:"
)
summary_chain = LLMChain(llm=llm, prompt=summary_prompt, output_key="summary")

# Combine into sequential chain
overall_chain = SequentialChain(
    chains=[extract_chain, analyze_chain, summary_chain],
    input_variables=["text"],
    output_variables=["entities", "analysis", "summary"],
    verbose=True
)

メモリ管理のベストプラクティス

適切なメモリタイプの選択

# For short conversations (< 10 messages)
from langchain.memory import ConversationBufferMemory
memory = ConversationBufferMemory()

# For long conversations (summarize old messages)
from langchain.memory import ConversationSummaryMemory
memory = ConversationSummaryMemory(llm=llm)

# For sliding window (last N messages)
from langchain.memory import ConversationBufferWindowMemory
memory = ConversationBufferWindowMemory(k=5)

# For entity tracking
from langchain.memory import ConversationEntityMemory
memory = ConversationEntityMemory(llm=llm)

# For semantic retrieval of relevant history
from langchain.memory import VectorStoreRetrieverMemory
memory = VectorStoreRetrieverMemory(retriever=retriever)

コールバックシステム

カスタムコールバックハンドラー

from langchain.callbacks.base import BaseCallbackHandler

class CustomCallbackHandler(BaseCallbackHandler):
    def on_llm_start(self, serialized, prompts, **kwargs):
        print("LLM started")

    def on_llm_end(self, response, **kwargs):
        print("LLM ended")

    def on_llm_error(self, error, **kwargs):
        print(f"LLM error: {error}")

    def on_chain_start(self, serialized, inputs, **kwargs):
        print("Chain started")

    def on_agent_action(self, action, **kwargs):
        print("Agent action selected")

# Use callback
agent.run("query", callbacks=[CustomCallbackHandler()])

テスト戦略

import pytest
from unittest.mock import Mock

def test_agent_tool_selection():
    # Mock LLM to return specific tool selection
    mock_llm = Mock()
    mock_llm.predict.return_value = "Action: search_database\nAction Input: test query"

    agent = initialize_agent(tools, mock_llm, agent=AgentType.ZERO_SHOT_REACT_DESCRIPTION)

    result = agent.run("test query")

    # Verify correct tool was selected
    # Assert the actual selected tool invocation with a compatible fake model/tool spy;
    # call arguments to the model only show the prompt, not tool execution.

def test_memory_persistence():
    memory = ConversationBufferMemory()

    memory.save_context({"input": "Hi"}, {"output": "Hello!"})

    assert "Hi" in memory.load_memory_variables({})['history']
    assert "Hello!" in memory.load_memory_variables({})['history']

パフォーマンスの最適化

1. キャッシュ

from langchain.cache import InMemoryCache
import langchain

langchain.llm_cache = InMemoryCache()

2. バッチ処理

# Process multiple documents in parallel
from langchain.document_loaders import DirectoryLoader
from concurrent.futures import ThreadPoolExecutor

loader = DirectoryLoader('./docs')
docs = loader.load()

def process_doc(doc):
    return text_splitter.split_documents([doc])

with ThreadPoolExecutor(max_workers=4) as executor:
    split_docs = list(executor.map(process_doc, docs))

3. ストリーミング応答

from langchain.callbacks.streaming_stdout import StreamingStdOutCallbackHandler

llm = OpenAI(streaming=True, callbacks=[StreamingStdOutCallbackHandler()])

リソース

• references/agents.md：エージェントアーキテクチャの詳細
• references/memory.md：メモリシステムのパターン
• references/chains.md：チェーン構成の戦略
• references/document-processing.md：ドキュメントの読み込みとインデックス作成
• references/callbacks.md：モニタリングと可観測性
• assets/agent-template.py：本番環境向けのエージェントテンプレート
• assets/memory-config.yaml：メモリ設定の例
• assets/chain-example.py：複雑なチェーンの例

よくある落とし穴

1. メモリのあふれ：会話履歴の長さを管理しない
2. ツール選択の誤り：ツールの説明が不適切だと、エージェントが混乱します
3. コンテキストウィンドウ超過：LLM のトークン上限を超過する
4. エラー処理がない：エージェントの失敗を捕捉して処理しない
5. 非効率な検索：ベクトルストアへのクエリを最適化しない

本番環境向けチェックリスト

• 適切なエラー処理を実装する
• リクエスト／レスポンスのログ記録を追加する
• トークンの使用量とコストを監視する
• エージェントの実行にタイムアウト制限を設定する
• レート制限を実装する
• 入力検証を追加する
• エッジケースをテストする
• 可観測性を設定する（コールバック）
• フォールバック戦略を実装する
• プロンプトと設定をバージョン管理する

制限事項

• 上記で説明した範囲にタスクが明確に当てはまる場合に限り、このスキルを使用してください。
• 出力を、環境固有の検証、テスト、専門家によるレビューの代替として扱わないでください。
• 必要な入力、権限、安全上の境界、または成功基準が不足している場合は、作業を中断して確認を求めてください。