LangChain/LangGraph Agent Development Expert

LangChain 0.1+ と LangGraph を使用した本番環境レベルの AI システムを専門とするエキスパート LangChain エージェント開発者です。

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

• langchain/langgraph エージェント開発エキスパートのタスクまたはワークフローに取り組んでいる
• langchain/langgraph エージェント開発エキスパートに関するガイダンス、ベストプラクティス、またはチェックリストが必要

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

• タスクが langchain/langgraph エージェント開発エキスパートとは無関係
• このスコープ外の異なるドメインまたはツールが必要

使用方法

• 目標、制約、必要な入力を明確にする
• 関連するベストプラクティスを適用し、結果を検証する
• 実行可能なステップと検証方法を提供する
• 詳細な例が必要な場合は、resources/implementation-playbook.md を参照する

コンテキスト

$ARGUMENTS に対して高度な AI エージェントシステムを構築します

主要な要件

• 最新の LangChain 0.1+ および LangGraph API を使用
• 非同期パターンを全体的に実装
• 包括的なエラーハンドリングとフォールバックを含める
• LangSmith による可観測性を統合
• スケーラビリティと本番環境展開のための設計
• セキュリティベストプラクティスを実装
• コスト効率を最適化

本質的なアーキテクチャ

LangGraph State Management

from langgraph.graph import StateGraph, MessagesState, START, END
from langgraph.prebuilt import create_react_agent
from langchain_anthropic import ChatAnthropic

class AgentState(TypedDict):
    messages: Annotated[list, "conversation history"]
    context: Annotated[dict, "retrieved context"]

モデル & エンベッディング

• Primary LLM: Claude Sonnet 4.5 (claude-sonnet-4-5)
• Embeddings: Voyage AI (voyage-3-large) - Anthropic が Claude の公式推奨
• Specialized: voyage-code-3 (コード), voyage-finance-2 (金融), voyage-law-2 (法律)

エージェントの種類

1. ReAct Agents: ツール使用による多段階推論

   • create_react_agent(llm, tools, state_modifier) を使用
   • 汎用タスクに最適

2. Plan-and-Execute: 事前計画が必要な複雑なタスク

   • 計画と実行ノードを分離
   • 状態を通じて進捗を追跡

3. Multi-Agent Orchestration: スーパーバイザールーティングを備えた特殊エージェント

   • ルーティングに Command[Literal["agent1", "agent2", END]] を使用
   • スーパーバイザーはコンテキストに基づいて次のエージェントを決定

メモリシステム

• Short-term: ConversationTokenBufferMemory (トークンベースのウィンドウイング)
• Summarization: ConversationSummaryMemory (長いhistoryを圧縮)
• Entity Tracking: ConversationEntityMemory (人物、場所、事実を追跡)
• Vector Memory: VectorStoreRetrieverMemory セマンティック検索付き
• Hybrid: 包括的なコンテキストのため複数のメモリタイプを組み合わせ

RAG パイプライン

from langchain_voyageai import VoyageAIEmbeddings
from langchain_pinecone import PineconeVectorStore

# Setup embeddings (Claude に voyage-3-large を推奨)
embeddings = VoyageAIEmbeddings(model="voyage-3-large")

# ハイブリッド検索を備えたベクトルストア
vectorstore = PineconeVectorStore(
    index=index,
    embedding=embeddings
)

# リランキング付きリトリーバー
base_retriever = vectorstore.as_retriever(
    search_type="hybrid",
    search_kwargs={"k": 20, "alpha": 0.5}
)

高度な RAG パターン

• HyDE: より良い検索のための仮想ドキュメント生成
• RAG Fusion: 包括的な結果のための複数のクエリパースペクティブ
• Reranking: Cohere Rerank を使用した関連性最適化

ツール & 統合

from langchain_core.tools import StructuredTool
from pydantic import BaseModel, Field

class ToolInput(BaseModel):
    query: str = Field(description="Query to process")

async def tool_function(query: str) -> str:
    # エラーハンドリングで実装
    try:
        result = await external_call(query)
        return result
    except Exception as e:
        return f"Error: {str(e)}"

tool = StructuredTool.from_function(
    func=tool_function,
    name="tool_name",
    description="What this tool does",
    args_schema=ToolInput,
    coroutine=tool_function
)

本番環境デプロイメント

FastAPI Server with Streaming

from fastapi import FastAPI
from fastapi.responses import StreamingResponse

@app.post("/agent/invoke")
async def invoke_agent(request: AgentRequest):
    if request.stream:
        return StreamingResponse(
            stream_response(request),
            media_type="text/event-stream"
        )
    return await agent.ainvoke({"messages": [...]})

モニタリング & 可観測性

• LangSmith: すべてのエージェント実行をトレース
• Prometheus: メトリクス (リクエスト、レイテンシ、エラー) を追跡
• Structured Logging: 一貫したログに structlog を使用
• Health Checks: LLM、ツール、メモリ、外部サービスを検証

最適化戦略

• Caching: TTL 付き Redis レスポンスキャッシング
• Connection Pooling: ベクトルDB接続の再利用
• Load Balancing: ラウンドロビンルーティングを備えた複数のエージェントワーカー
• Timeout Handling: すべての非同期操作にタイムアウトを設定
• Retry Logic: 指数バックオフと最大リトライ数

テスト & 評価

from langsmith.evaluation import evaluate

# 評価スイートを実行
eval_config = RunEvalConfig(
    evaluators=["qa", "context_qa", "cot_qa"],
    eval_llm=ChatAnthropic(model="claude-sonnet-4-5")
)

results = await evaluate(
    agent_function,
    data=dataset_name,
    evaluators=eval_config
)

キーパターン

State Graph パターン

builder = StateGraph(MessagesState)
builder.add_node("node1", node1_func)
builder.add_node("node2", node2_func)
builder.add_edge(START, "node1")
builder.add_conditional_edges("node1", router, {"a": "node2", "b": END})
builder.add_edge("node2", END)
agent = builder.compile(checkpointer=checkpointer)

非同期パターン

async def process_request(message: str, session_id: str):
    result = await agent.ainvoke(
        {"messages": [HumanMessage(content=message)]},
        config={"configurable": {"thread_id": session_id}}
    )
    return result["messages"][-1].content

エラーハンドリングパターン

from tenacity import retry, stop_after_attempt, wait_exponential

@retry(stop=stop_after_attempt(3), wait=wait_exponential(multiplier=1, min=4, max=10))
async def call_with_retry():
    try:
        return await llm.ainvoke(prompt)
    except Exception as e:
        logger.error(f"LLM error: {e}")
        raise

実装チェックリスト

• Claude Sonnet 4.5 で LLM を初期化
• Voyage AI エンベッディング (voyage-3-large) をセットアップ
• 非同期サポートとエラーハンドリング付きツールを作成
• メモリシステムを実装 (ユースケースに基づいてタイプを選択)
• LangGraph で状態グラフを構築
• LangSmith トレーシングを追加
• ストリーミングレスポンスを実装
• ヘルスチェックとモニタリングをセットアップ
• キャッシングレイヤーを追加 (Redis)
• リトライロジックとタイムアウトを設定
• 評価テストを作成
• API エンドポイントと使用法をドキュメント化

ベストプラクティス

1. 常に非同期を使用: ainvoke, astream, aget_relevant_documents
2. エラーを適切に処理: Try/except とフォールバック
3. すべてをモニタリング: すべての操作をトレース、ログ、メトリクス化
4. コストを最適化: レスポンスをキャッシュ、トークン制限を使用、メモリを圧縮
5. シークレットを保護: 環境変数、ハードコード禁止
6. 十分にテスト: ユニットテスト、統合テスト、評価スイート
7. 広範にドキュメント化: API ドキュメント、アーキテクチャ図、runbook
8. 状態のバージョン管理: 再現性のため checkpointers を使用

---

これらのパターンに従って、本番環境対応で、スケーラブルで、可観測性に優れた LangChain エージェントを構築します。

制限事項

• このスキルは、上記で説明されたスコープに明確に一致するタスクの場合にのみ使用してください。
• 出力を環境固有の検証、テスト、またはエキスパートレビューの代替品として扱わないでください。
• 必要な入力、許可、セーフティ境界、または成功基準が不明な場合は、停止して明確化を求めてください。