langchain-architecture

LangChain & LangGraph アーキテクチャ

最新の LangChain 1.x と LangGraph をマスターし、エージェント、状態管理、メモリ、ツール統合を備えた高度な LLM アプリケーションを構築します。

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

• ツールにアクセスできる自律型 AI エージェントの構築
• 複雑な多段階 LLM ワークフローの実装
• 会話メモリと状態の管理
• LLM と外部データソースおよび API の統合
• モジュール式で再利用可能な LLM アプリケーションコンポーネントの作成
• ドキュメント処理パイプラインの実装
• 本番グレードの LLM アプリケーションの構築

パッケージ構成 (LangChain 1.x)

langchain (1.2.x)         # High-level orchestration
langchain-core (1.2.x)    # Core abstractions (messages, prompts, tools)
langchain-community       # Third-party integrations
langgraph                 # Agent orchestration and state management
langchain-openai          # OpenAI integrations
langchain-anthropic       # Anthropic/Claude integrations
langchain-voyageai        # Voyage AI embeddings
langchain-pinecone        # Pinecone vector store

中核となる概念

1. LangGraph エージェント

LangGraph は 2026 年におけるエージェント構築の標準です。以下を提供します。

主な機能:

• StateGraph: 型付き状態による明示的な状態管理
• 耐久性のある実行: エージェントは障害を経ても持続する
• Human-in-the-Loop: 任意の時点で状態を検査および変更
• メモリ: セッションをまたぐ短期メモリと長期メモリ
• チェックポイント: エージェントの状態を保存して再開

エージェントパターン:

• ReAct: create_react_agent による推論 + 行動
• Plan-and-Execute: 計画ノードと実行ノードの分離
• マルチエージェント: 特化したエージェント間のスーパーバイザールーティング
• ツール呼び出し: Pydantic スキーマによる構造化されたツール呼び出し

2. 状態管理

LangGraph は明示的な状態のために TypedDict を使用します:

from typing import Annotated, TypedDict
from langgraph.graph import MessagesState

# Simple message-based state
class AgentState(MessagesState):
    """Extends MessagesState with custom fields."""
    context: Annotated[list, "retrieved documents"]

# Custom state for complex agents
class CustomState(TypedDict):
    messages: Annotated[list, "conversation history"]
    context: Annotated[dict, "retrieved context"]
    current_step: str
    results: list

3. メモリシステム

最新のメモリ実装:

• ConversationBufferMemory: すべてのメッセージを保存 (短い会話)
• ConversationSummaryMemory: 古いメッセージを要約 (長い会話)
• ConversationTokenBufferMemory: トークンベースのウィンドウ処理
• VectorStoreRetrieverMemory: 意味的類似性による検索
• LangGraph チェックポインター: セッションをまたぐ永続状態

4. ドキュメント処理

ドキュメントの読み込み、変換、保存:

コンポーネント:

• ドキュメントローダー: さまざまなソースから読み込み
• テキストスプリッター: ドキュメントをインテリジェントにチャンク化
• ベクトルストア: 埋め込みの保存と検索
• レトリーバー: 関連ドキュメントの取得

5. コールバックとトレーシング

LangSmith はオブザーバビリティの標準です:

• リクエスト/レスポンスのログ記録
• トークン使用量の追跡
• レイテンシの監視
• エラーの追跡
• トレースの可視化

クイックスタート

LangGraph による最新の ReAct エージェント

from langgraph.prebuilt import create_react_agent
from langgraph.checkpoint.memory import MemorySaver
from langchain_anthropic import ChatAnthropic
from langchain_core.tools import tool
import ast
import operator

# Initialize LLM (Claude Sonnet 5 recommended)
llm = ChatAnthropic(model="claude-sonnet-5")

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

@tool
def calculate(expression: str) -> str:
    """Safely evaluate a mathematical expression.

    Supports: +, -, *, /, **, %, parentheses
    Example: '(2 + 3) * 4' returns '20'
    """
    # Safe math evaluation using ast
    allowed_operators = {
        ast.Add: operator.add,
        ast.Sub: operator.sub,
        ast.Mult: operator.mul,
        ast.Div: operator.truediv,
        ast.Pow: operator.pow,
        ast.Mod: operator.mod,
        ast.USub: operator.neg,
    }

    def _eval(node):
        if isinstance(node, ast.Constant):
            return node.value
        elif isinstance(node, ast.BinOp):
            left = _eval(node.left)
            right = _eval(node.right)
            return allowed_operators[type(node.op)](left, right)
        elif isinstance(node, ast.UnaryOp):
            operand = _eval(node.operand)
            return allowed_operators[type(node.op)](operand)
        else:
            raise ValueError(f"Unsupported operation: {type(node)}")

    try:
        tree = ast.parse(expression, mode='eval')
        return str(_eval(tree.body))
    except Exception as e:
        return f"Error: {e}"

tools = [search_database, calculate]

# Create checkpointer for memory persistence
checkpointer = MemorySaver()

# Create ReAct agent
agent = create_react_agent(
    llm,
    tools,
    checkpointer=checkpointer
)

# Run agent with thread ID for memory
config = {"configurable": {"thread_id": "user-123"}}
result = await agent.ainvoke(
    {"messages": [("user", "Search for Python tutorials and calculate 25 * 4")]},
    config=config
)

詳細なパターンと実例

詳細なパターンのドキュメントは references/details.md にあります。上記のナビゲーション階層では不十分な場合は、そのファイルをお読みください。

テスト戦略

import pytest
from unittest.mock import AsyncMock, patch

@pytest.mark.asyncio
async def test_agent_tool_selection():
    """Test agent selects correct tool."""
    with patch.object(llm, 'ainvoke') as mock_llm:
        mock_llm.return_value = AsyncMock(content="Using search_database")

        result = await agent.ainvoke({
            "messages": [("user", "search for documents")]
        })

        # Verify tool was called
        assert "search_database" in str(result)

@pytest.mark.asyncio
async def test_memory_persistence():
    """Test memory persists across invocations."""
    config = {"configurable": {"thread_id": "test-thread"}}

    # First message
    await agent.ainvoke(
        {"messages": [("user", "Remember: the code is 12345")]},
        config
    )

    # Second message should remember
    result = await agent.ainvoke(
        {"messages": [("user", "What was the code?")]},
        config
    )

    assert "12345" in result["messages"][-1].content

パフォーマンスの最適化

1. Redis によるキャッシュ

from langchain_community.cache import RedisCache
from langchain_core.globals import set_llm_cache
import redis

redis_client = redis.Redis.from_url("redis://localhost:6379")
set_llm_cache(RedisCache(redis_client))

2. 非同期バッチ処理

import asyncio
from langchain_core.documents import Document

async def process_documents(documents: list[Document]) -> list:
    """Process documents in parallel."""
    tasks = [process_single(doc) for doc in documents]
    return await asyncio.gather(*tasks)

async def process_single(doc: Document) -> dict:
    """Process a single document."""
    chunks = text_splitter.split_documents([doc])
    embeddings = await embeddings_model.aembed_documents(
        [c.page_content for c in chunks]
    )
    return {"doc_id": doc.metadata.get("id"), "embeddings": embeddings}

3. コネクションプーリング

from langchain_pinecone import PineconeVectorStore
from pinecone import Pinecone

# Reuse Pinecone client
pc = Pinecone(api_key=os.environ["PINECONE_API_KEY"])
index = pc.Index("my-index")

# Create vector store with existing index
vectorstore = PineconeVectorStore(index=index, embedding=embeddings)