llm-application-dev-langchain-agent

LangChain/LangGraph エージェント開発のエキスパート

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

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

• langchain/langgraph のエージェント開発に関するエキスパートタスクやワークフローに取り組む場合
• langchain/langgraph のエージェント開発について、ガイダンス、ベストプラクティス、チェックリストが必要な場合

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

• タスクが langchain/langgraph のエージェント開発に関するものではない場合
• この範囲外の別のドメインやツールが必要な場合

手順

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

コンテキスト

次の用途向けに、高度な AI エージェントシステムを構築します：$ARGUMENTS

主な要件

• 互換性のある LangChain/LangGraph/プロバイダーのバージョンを選び、固定してください。レガシー API と v1 API を混在させないでください。
• 全体を通して非同期パターンを実装してください。
• 包括的なエラー処理とフォールバックを含めてください。
• 可観測性のために LangSmith を統合してください。
• スケーラビリティと本番環境へのデプロイを考慮して設計してください。
• セキュリティのベストプラクティスを実装してください。
• コスト効率を最適化してください。

基本アーキテクチャ

LangGraph の状態管理

from langgraph.graph import StateGraph, MessagesState, START, END
from langchain.agents import create_agent
from langchain_anthropic import ChatAnthropic

from typing import TypedDict, Annotated
from langgraph.graph.message import add_messages

class AgentState(TypedDict):
    messages: Annotated[list, add_messages]
    context: dict

モデルと埋め込み

• LLM の例：Claude Sonnet 4.5（claude-sonnet-4-5）
• 埋め込み：Voyage AI（voyage-3-large）— モデルの提供状況と、固定したインテグレーションとの適合性を確認してください。
• 特化型：voyage-code-3（コード）、voyage-finance-2（金融）、voyage-law-2（法律）

エージェントの種類

1. ReAct エージェント：ツールを使用した複数ステップの推論

   • LangChain v1 では create_agent(model=llm, tools=tools, system_prompt="...") を使用します。
   • 汎用タスクに最適です。

2. 計画・実行型：事前の計画が必要な複雑なタスク

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

3. マルチエージェントオーケストレーション：スーパーバイザーによるルーティングを行う特化型エージェント

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

メモリシステム

• 短期：ConversationTokenBufferMemory（トークンベースのウィンドウ化）
• 要約：ConversationSummaryMemory（長い履歴を圧縮）
• エンティティ追跡：ConversationEntityMemory（人物、場所、事実を追跡）
• ベクトルメモリ：VectorStoreRetrieverMemory とセマンティック検索
• ハイブリッド：複数のメモリタイプを組み合わせて、包括的なコンテキストを実現

RAG パイプライン

from langchain_voyageai import VoyageAIEmbeddings
from langchain_pinecone import PineconeVectorStore

# Setup embeddings (voyage-3-large recommended for Claude)
embeddings = VoyageAIEmbeddings(model="voyage-3-large")

# Vector store with dense similarity search
vectorstore = PineconeVectorStore(
    index=index,
    embedding=embeddings
)

# Dense retriever; add a separately implemented reranker if needed
base_retriever = vectorstore.as_retriever(
    search_type="similarity",
    search_kwargs={"k": 20}
)

高度な RAG パターン

• HyDE：検索精度を高めるため、仮説上のドキュメントを生成します。
• RAG Fusion：包括的な結果を得るため、複数のクエリ視点を使用します。
• 再ランキング：関連性を最適化するため、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:
    # Implement with error handling
    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 サーバー

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：メトリクス（リクエスト数、レイテンシ、エラー）を追跡します。
• 構造化ログ：一貫性のあるログを記録するために structlog を使用します。
• ヘルスチェック：LLM、ツール、メモリ、外部サービスを検証します。

最適化戦略

• キャッシュ：TTL を設定したレスポンスキャッシュに Redis を使用します。
• コネクションプーリング：ベクトル DB の接続を再利用します。
• ロードバランシング：ラウンドロビン方式のルーティングで複数のエージェントワーカーを使用します。
• タイムアウト処理：すべての非同期処理にタイムアウトを設定します。
• リトライロジック：最大リトライ回数を設定した指数バックオフを使用します。

テストと評価

from langsmith.evaluation import evaluate

# Prerequisites: a real callable agent_function, dataset_name and a list of
# compatible evaluator callables; supply and test them for the pinned LangSmith SDK.
results = evaluate(agent_function, data=dataset_name, evaluators=evaluator_functions)

主なパターン

状態グラフパターン

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 ドキュメント、アーキテクチャ図、運用手順書
8. 状態のバージョンを管理する：再現性のためにチェックポインターを使用する

---

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

制限事項

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

例は部分的なスキーマ／断片であり、従来のメモリー名には langchain-classic が必要な場合があります。これらは、そのまま実行できる本番用エージェントではありません。現在の v1 では create_agent と system_prompt を使用します。公式の移行ガイドを確認してください。認証済みアカウントで実際に利用できるモデル ID を指定し、リトライとコストに上限を設け、ツールの引数とユーザー権限を検証してください。チェックポイントのスレッド ID は認証済みのテナント／ユーザーの範囲に限定し、非公開のトレースエクスポートへのアクセスを認可してください。呼び出し元が指定したスレッド ID はアクセス確認にはなりません。ツールのエラーは機密情報を伏せて返し、非公開の内部推論ではなく、簡潔な推論の要約を返してください。