LangGraphを使ったLangChain Agentの思考フローの可視化

from typing import TypedDict, Literal

from langgraph.graph import StateGraph, END

from my_agent.utils.nodes import call_model, should_continue, tool_node
from my_agent.utils.state import AgentState

# Define the config
class GraphConfig(TypedDict):
    model_name: Literal["anthropic", "openai"]

# Define a new graph
workflow = StateGraph(AgentState, config_schema=GraphConfig)

# Define the two nodes we will cycle between
workflow.add_node("agent", call_model)
workflow.add_node("action", tool_node)

# Set the entrypoint as `agent`
# This means that this node is the first one called
workflow.set_entry_point("agent")

# We now add a conditional edge
workflow.add_conditional_edges(
    # First, we define the start node. We use `agent`.
    # This means these are the edges taken after the `agent` node is called.
    "agent",
    # Next, we pass in the function that will determine which node is called next.
    should_continue,
    # Finally we pass in a mapping.
    # The keys are strings, and the values are other nodes.
    # END is a special node marking that the graph should finish.
    # What will happen is we will call `should_continue`, and then the output of that
    # will be matched against the keys in this mapping.
    # Based on which one it matches, that node will then be called.
    {
        # If `tools`, then we call the tool node.
        "continue": "action",
        # Otherwise we finish.
        "end": END,
    },
)

# We now add a normal edge from `tools` to `agent`.
# This means that after `tools` is called, `agent` node is called next.
workflow.add_edge("action", "agent")

# Finally, we compile it!
# This compiles it into a LangChain Runnable,
# meaning you can use it as you would any other runnable
graph = workflow.compile()

{
  "dependencies": ["."],
  "graphs": {
    "agent": "./my_agent/agent.py:graph",
    "debug_agent": "./my_agent/debug_agent.py:debug_agent_graph"
  },
  "env": ".env"
}

pip install langchain langgraph langchain-openai langchain-anthropic python-dotenv

langchain: 0.2.17
langgraph: 0.2.76
langchain-openai: 0.1.25
langchain-anthropic: 0.1.23
python-dotenv: 1.0.1

OPENAI_API_KEY=your_openai_api_key
TAVILY_API_KEY=your_tavily_api_key

pip install langgraph[cli]

langgraph dev

import os
import random
import time
from typing import Dict, List, Any, Optional, TypedDict
from dotenv import load_dotenv
from langchain_openai import ChatOpenAI
from langchain_core.messages import HumanMessage, AIMessage, SystemMessage
from langgraph.graph import StateGraph, END, START

# 環境変数の読み込み
load_dotenv()

print("LangGraph デバッグサンプルを開始します...")

# デバッグ設定
DEBUG = True

# 状態の型定義
class DebugState(TypedDict):
    """デバッググラフの状態を表現するクラス"""
    messages: List[Any]  # メッセージ履歴
    current_step: str    # 現在のステップ
    attempts: int        # 試行回数
    has_error: bool      # エラーが発生したかどうか
    error_message: Optional[str]  # エラーメッセージ
    result: Dict[str, Any]  # 結果
    force_error_search: bool  # 検索エラーを強制するフラグ
    force_error_calculate: bool  # 計算エラーを強制するフラグ

# ノード関数の定義
def initial_node(state: DebugState) -> DebugState:
    """初期ノード: ユーザーの入力を処理する"""
    print("初期ノードを実行中...")
    
    messages = state["messages"]
    if not messages:
        # メッセージがない場合はデフォルトメッセージを追加
        messages = [HumanMessage(content="AIについて調べてください")]
    
    # JSON形式のメッセージを適切なメッセージオブジェクトに変換
    processed_messages = []
    for msg in messages:
        if isinstance(msg, dict):
            # 辞書形式のメッセージをオブジェクトに変換
            if msg.get("type") == "human":
                processed_messages.append(HumanMessage(content=msg.get("content", "")))
            elif msg.get("type") == "ai":
                processed_messages.append(AIMessage(content=msg.get("content", "")))
            elif msg.get("type") == "system":
                processed_messages.append(SystemMessage(content=msg.get("content", "")))
            else:
                # 不明なタイプの場合はヒューマンメッセージとして扱う
                processed_messages.append(HumanMessage(content=str(msg.get("content", ""))))
        else:
            # すでにメッセージオブジェクトの場合はそのまま追加
            processed_messages.append(msg)
    
    # 処理済みメッセージに置き換え
    messages = processed_messages
    
    # LLMを使ってプロンプト内容を分析
    llm = ChatOpenAI(model="gpt-4o", temperature=0)
    system_message = SystemMessage(content="""
    あなたはプロンプト分析アシスタントです。ユーザーの質問を分析し、必要な処理ステップを特定してください。
    選択肢:
    - search: 情報検索が必要な場合
    - calculate: 数値計算や予測が必要な場合
    
    回答は「search」「calculate」のいずれかのみを返してください。
    """)
    
    # プロンプト内容を分析して次のステップを決定
    action_input = messages[-1].content if messages else "AIについて調べてください"
    action_prompt = f"次のユーザープロンプトを分析し、最適な処理ステップを選択してください: {action_input}"
    
    response = llm.invoke([system_message, HumanMessage(content=action_prompt)])
    next_step = response.content.strip().lower()
    
    # 有効なステップに正規化
    valid_steps = ["search", "calculate"]
    if next_step not in valid_steps:
        next_step = "search"  # デフォルト
    
    # 解析結果をメッセージに追加
    messages.append(AIMessage(content=f"プロンプト分析: '{next_step}' 処理が必要と判断しました。"))
    
    # エラーフラグをチェック（メッセージに特定のキーワードがあるか）
    force_error_search = "エラー" in action_input or "失敗" in action_input or "error" in action_input.lower()
    force_error_calculate = "計算エラー" in action_input or "計算失敗" in action_input or "calculation error" in action_input.lower()
    
    return {
        **state,
        "messages": messages,
        "current_step": next_step,
        "attempts": 0,
        "has_error": False,
        "error_message": None,
        "force_error_search": force_error_search,
        "force_error_calculate": force_error_calculate
    }

def search_node(state: DebugState) -> DebugState:
    """検索ノード: 情報検索を行う"""
    print("検索ノードを実行中...")
    
    # 検索処理のシミュレーション
    time.sleep(1)
    
    # 検索結果をメッセージに追加
    messages = state["messages"]
    messages.append(AIMessage(content="検索結果: AIに関する最新情報を見つけました。"))
    
    # エラーを強制するパラメータをチェック
    force_error = state.get("force_error_search", False)
    
    # エラーが強制されるか、デバッグモードでランダムにエラーが発生する場合
    if force_error or (random.random()  DebugState:
    """計算ノード: 数値計算を行う"""
    print("計算ノードを実行中...")
    
    # 計算処理のシミュレーション
    time.sleep(1)
    
    # 計算結果をメッセージに追加
    messages = state["messages"]
    messages.append(AIMessage(content="計算結果: AIの成長率は年間約70%と推定されます。"))
    
    # エラーを強制するパラメータをチェック
    force_error = state.get("force_error_calculate", False)
    
    # エラーが強制されるか、デバッグモードでランダムにエラーが発生する場合
    if force_error or (random.random()  DebugState:
    """エラーハンドリングノード: エラーを処理する"""
    print("エラーハンドリングノードを実行中...")
    
    messages = state["messages"]
    error_message = state.get("error_message", "不明なエラーが発生しました")
    attempts = state.get("attempts", 0) + 1
    
    # エラーメッセージをログに記録
    print(f"エラー処理 (試行 {attempts}): {error_message}")
    
    # エラーメッセージをメッセージに追加
    messages.append(AIMessage(content=f"エラーが発生しました: {error_message}"))
    
    # 再試行するかどうかの判断
    if attempts  DebugState:
    """最終ノード: 最終結果を生成する"""
    print("最終ノードを実行中...")
    
    messages = state["messages"]
    has_error = state.get("has_error", False)
    
    if has_error:
        messages.append(AIMessage(content="エラーが発生したため、完全な結果を提供できませんでした。利用可能な情報に基づいて回答します。"))
    
    # LLMを使って最終回答を生成
    llm = ChatOpenAI(model="gpt-4o", temperature=0)
    system_message = SystemMessage(content="あなたはデバッグテスト用のアシスタントです。収集した情報に基づいて、最終的な回答を提供してください。")
    
    # これまでのメッセージ履歴に基づいて回答を生成
    final_response = llm.invoke([system_message] + messages)
    
    # 最終回答をメッセージに追加
    messages.append(final_response)
    
    # 結果をまとめる
    result = {
        "final_answer": final_response.content,
        "error_occurred": has_error,
        "attempt_count": state.get("attempts", 0)
    }
    
    return {
        **state,
        "messages": messages,
        "current_step": "end",
        "result": result
    }

# ルーター関数
def router_node(state: DebugState) -> str:
    """次に実行するノードを決定する"""
    current_step = state.get("current_step", "")
    has_error = state.get("has_error", False)
    
    print(f"ルーター: 現在のステップ = {current_step}, エラー = {has_error}")
    
    if current_step == "end":
        return "end"
    
    return current_step

# グラフ作成関数
def create_debug_graph():
    """デバッググラフを作成"""
    # ワークフローの定義
    workflow = StateGraph(DebugState)
    
    # ノードの追加
    workflow.add_node("initial", initial_node)
    workflow.add_node("search", search_node)
    workflow.add_node("calculate", calculate_node)
    workflow.add_node("error", error_node)
    workflow.add_node("final", final_node)
    
    # エントリーポイントの設定
    workflow.add_edge(START, "initial")
    
    # 初期ノードからの条件付きエッジ
    workflow.add_conditional_edges(
        "initial",
        router_node,
        {
            "search": "search",
            "calculate": "calculate",
            "error": "error",
            "final": "final"
        }
    )
    
    # 各処理ノードからの条件付きエッジ
    for node in ["search", "calculate", "error"]:
        workflow.add_conditional_edges(
            node,
            router_node,
            {
                "search": "search",
                "calculate": "calculate",
                "error": "error",
                "final": "final"
            }
        )
    
    # 最終ノードからENDへのエッジのみを残す
    workflow.add_conditional_edges(
        "final",
        router_node,
        {
            "end": END
        }
    )
    
    # グラフのコンパイル
    return workflow.compile()

# グラフのインスタンス化
debug_agent_graph = create_debug_graph()

{
  "messages": [
    {
      "type": "human",
      "content": "AIの成長率を計算して将来の予測をしてください"
    }
  ]
}

{
  "messages": [
    {
      "type": "human",
      "content": "AIについて調べてエラーを発生させてください"
    }
  ]
}