LangGraphとBox APIによるAIを活用したワークフローの構築

大規模言語モデル (LLM) の機能と構造化されたワークフローやクラウドストレージAPIを組み合わせることで、クラウドに保存されたドキュメントを分析、処理し、インサイトを生成できるインテリジェントなシステムを作成できます。

このワークショップでは、LangGraphとBox APIを使用してAIを活用したワークフローを構築する方法について説明します。Boxに保存されている映画脚本を処理して、その内容を分析し、登場人物、場所、小道具などの詳細を含む包括的なレポートを生成するサンプルアプリケーションを作成します。

この例に関する次の動画もご覧ください。

説明する内容

ドキュメントを処理するLangGraphワークフローの設定方法
Box APIとAIエージェントを統合する方法
さまざまなワークフローパターン (順次、並列、条件付き)
AIエージェント用のBox APIツールの使用方法
ワークフローで状態を構造化して整理する方法
AIを活用した堅牢なワークフローを構築するためのベストプラクティス

LangGraphについて

LangGraphは、LLMを使用してステートフルなマルチアクターアプリケーションを構築するためのライブラリです。これを使用すると、以下のことが可能です。

さまざまな状態や遷移を使用して構造化されたワークフローを定義する
ワークフローのステップ間で状態を維持する
条件分岐、並列実行などを使用した複雑なワークフローを作成する
さまざまなツールやAPIと統合する

LangGraphの中核となる概念はStateGraphで、ワークフローを次のような有向グラフとして表します。

ノードは処理ステップ (状態を変換する関数) である
エッジはステップ間のフローを定義する
状態はあるステップから次のステップに渡される

Box APIの統合

Box APIを使用すると、Boxのコンテンツクラウドを操作できるため、AIワークフローで以下の処理を実行できます。

ファイルやフォルダを検索する
ファイルコンテンツを読み取る
Box AI機能 (データ抽出や質問への回答など) を使用する
フォルダのコンテンツのリストを取得する
ファイルやフォルダを作成、更新、削除する

box-ai-agents-toolkitパッケージを使用して、Box APIの統合を簡略化します。

ワークフローの設計

Boxに保存されている映画脚本を処理するワークフローを設計します。

Box内で脚本ファイルを探す
脚本の内容を読み取る
脚本を分析して主要な情報を調べる
脚本で言及されている場所
脚本における登場人物の役割
脚本で使用されている小道具
各登場人物の役に俳優を提案する
脚本の作者を分析する
適切なプロデューサーや監督を提案する
包括的なレポートを生成する

以下は、ワークフローを視覚的に表したものです。

0_AsJQ8xHDj4kDbOSo

データモデルの定義

まず、ワークフローの各ステップでデータを表すPydanticモデルを定義します。これらのモデルでは、ワークフローの状態を構造化し、型検証を行います。

# workflow_classes.py
from typing import TypedDict, Union
from pydantic import BaseModel, Field

class BoxFileLocation(BaseModel):
    """Model for Box file location."""
    file_name: str = Field(None, description="Name of the file.")
    file_id: str = Field(None, description="Box file id.")
    parent_folder_name: str = Field(None, description="Name of the parent folder.")

class ScriptData(BaseModel):
    """Model for script data."""
    title: str = Field(None, description="Title of the script.")
    author: str = Field(None, description="Author of the script.")
    genre: str = Field(None, description="Genre of the script.")
    date: str = Field(None, description="Date of the script.")
    plot_summary: str = Field(None, description="Plot summary of the script.")

# Additional models for Locations, Characters, Props, etc.
# ...

# Workflow state that combines all models
class WorkFlowState(TypedDict):
    box_script_file: BoxFileLocation
    script_file_read: str
    script_data: ScriptData
    locations: Locations
    roles: Roles
    characters: Characters
    props: Props
    author: Author
    producers: Producers
    directors: Directors
    markdown: str

Box APIツールの設定

次に、AIエージェントが使用するBox APIツールを設定します。

# box_agent_tools.py
from box_ai_agents_toolkit import (
    File, Folder, SearchForContentContentTypes,
    box_file_ai_ask, box_file_ai_extract, box_file_text_extract,
    box_folder_list_content, box_locate_folder_by_name,
    box_search, get_ccg_client
)
from langchain.tools.base import StructuredTool
from langchain_core.tools import BaseTool
from langgraph.prebuilt import create_react_agent

def init_tools():
    """Initialize the tools for the agent."""
    tools: List[BaseTool] = []
    
    # Add Box API tools
    tools.append(StructuredTool.from_function(box_who_am_i, parse_docstring=True))
    tools.append(StructuredTool.from_function(box_search_tool, parse_docstring=True))
    tools.append(StructuredTool.from_function(box_read_tool, parse_docstring=True))
    tools.append(StructuredTool.from_function(box_ask_ai_tool, parse_docstring=True))
    tools.append(StructuredTool.from_function(box_search_folder_by_name, parse_docstring=True))
    tools.append(StructuredTool.from_function(box_ai_extract_data, parse_docstring=True))
    tools.append(StructuredTool.from_function(box_list_folder_content_by_folder_id, parse_docstring=True))
    
    return tools

def get_box_agent(has_memory=False, response_format=None):
    """Create a Box agent with the specified tools."""
    model = init_chat_model("gpt-4o", model_provider="openai")
    tools = init_tools()
    
    memory = MemorySaver() if has_memory else None
    return create_react_agent(model, tools, checkpointer=memory, response_format=response_format)

# Individual tool implementations
def box_search_tool(query, file_extensions=None, where_to_look_for_query=None, ancestor_folder_ids=None):
    """Searches for files in Box using the specified query and filters."""
    client = get_ccg_client()
    # Convert search parameters and execute search
    # ...
    return [file.to_dict() for file in search_results]

# Additional tool implementations
# ...

ワークフローのステップの実装

ここで、ワークフローの個々のステップを、ワークフローの状態を変換する関数として実装します。

# workflow_steps.py
from workflow_classes import WorkFlowState, BoxFileLocation, ScriptData, Locations, Roles, Props, Characters, Author, Producers, Directors
from box_agent_tools import get_box_agent

def step_locate_file_in_box(state: WorkFlowState) -> WorkFlowState:
    """Locate a file in Box."""
    box_agent = get_box_agent(has_memory=False, response_format=BoxFileLocation)
    response = box_agent.invoke({
        "messages": [{
            "role": "user",
            "content": f"locate the file {state['box_script_file'].file_name} under my {state['box_script_file'].parent_folder_name} folder",
        }]
    })
    state["box_script_file"] = response["structured_response"]
    return state

def step_check_file(state: WorkFlowState) -> str:
    """Check if the file exists in Box."""
    if state["box_script_file"].file_id:
        return "Found"
    else:
        return "Not Found"

def step_read_box_file(state: WorkFlowState) -> WorkFlowState:
    """Read the file from Box."""
    box_agent = get_box_agent(has_memory=False)
    response = box_agent.invoke({
        "messages": [{
            "role": "user",
            "content": f"Read the box file id {state['box_script_file'].file_id} and respond with the actual text content of the movie script",
        }]
    })
    state["script_file_read"] = response["messages"][-1].content
    return state

# Additional step implementations
# ...

ワークフローグラフの作成

次に、ノード (ステップ) とエッジ (遷移) を定義してワークフローグラフを作成します。

# workflow_design.py
from langgraph.graph import END, START, StateGraph
from workflow_steps import (
    WorkFlowState, step_analyze_author, step_analyze_locations,
    step_analyze_props, step_analyze_roles, step_analyze_script,
    step_check_file, step_create_markdown, step_locate_file_in_box,
    step_potential_directors, step_potential_producers, step_read_box_file,
    step_suggest_actors_for_role
)

def build_workflow():
    """Builds the workflow for the script analysis process."""
    workflow = StateGraph(WorkFlowState)
    
    # Add nodes (workflow steps)
    workflow.add_node("locate_file_in_box", step_locate_file_in_box)
    workflow.add_node("read_box_file", step_read_box_file)
    workflow.add_node("analyze_script", step_analyze_script)
    workflow.add_node("analyze_locations", step_analyze_locations)
    workflow.add_node("analyze_roles", step_analyze_roles)
    workflow.add_node("analyze_props", step_analyze_props)
    workflow.add_node("suggest_actors_for_role", step_suggest_actors_for_role)
    workflow.add_node("analyze_author", step_analyze_author)
    workflow.add_node("potential_producers", step_potential_producers)
    workflow.add_node("potential_directors", step_potential_directors)
    workflow.add_node("create_markdown", step_create_markdown)
    
    # Add sequential edges
    workflow.add_edge(START, "locate_file_in_box")
    
    # Add conditional edge
    workflow.add_conditional_edges(
        "locate_file_in_box",
        step_check_file,
        {
            "Found": "read_box_file",
            "Not Found": END,
        },
    )
    
    workflow.add_edge("read_box_file", "analyze_script")
    
    # Add parallel execution paths
    workflow.add_edge("analyze_script", "analyze_locations")
    workflow.add_edge("analyze_script", "analyze_props")
    workflow.add_edge("analyze_script", "analyze_roles")
    workflow.add_edge("analyze_roles", "suggest_actors_for_role")
    
    # Join parallel paths
    workflow.add_edge(
        ["suggest_actors_for_role", "analyze_locations", "analyze_props"],
        "analyze_author",
    )
    
    # More parallel execution
    workflow.add_edge("analyze_author", "potential_producers")
    workflow.add_edge("analyze_author", "potential_directors")
    
    # Join and finish
    workflow.add_edge(["potential_producers", "potential_directors"], "create_markdown")
    workflow.add_edge("create_markdown", END)
    
    return workflow.compile()

ワークフローの実行

最後に、ワークフローを実行するスクリプトを作成します。

# demo.py
import dotenv
from box_ai_agents_toolkit import get_ccg_client
from workflow_design import build_workflow
from workflow_classes import BoxFileLocation, WorkFlowState
from file_utils import save_markdown

dotenv.load_dotenv()

def main():
    """Main function to run the demo."""
    # Initialize Box client
    client = get_ccg_client()
    user_info = client.users.get_user_me()
    print(f"Connected as: {user_info.name}")
    
    # Build and run the workflow
    chain = build_workflow()
    state = WorkFlowState(
        box_script_file=BoxFileLocation(
            file_name="Hitchhiker's-Guide-to-the-Galaxy-The",
            parent_folder_name="Scripts",
        )
    )
    
    # Invoke the workflow
    state = chain.invoke(state)
    
    # Check results and save
    if state["box_script_file"].file_id is None:
        print("File not found. Please check the file name and try again.")
        return
    
    # Save the markdown report
    save_markdown(
        state["markdown"],
        "output/script_analysis.md",
    )
    print("Analysis complete! Report saved to output/script_analysis.md")

if __name__ == "__main__":
    main()

LangGraphの主な機能

今回紹介したLangGraphの主な機能を以下に示します。

ワークフローの定義に使用したStateGraph

LangGraphのStateGraphを使用すると、ワークフローをノードとエッジを含む有向グラフとして定義できます。これにより、ワークフローが明確に視覚化され、データの流れについて簡単に推論できるようになります。

ワークフローパターン

デモとしていくつかの重要なワークフローパターンを示しました。

順次処理: 次から次へと続くステップ (例: ファイルを探す → ファイルを読み取る → 脚本を分析する)
条件分岐: 条件 (例: ファイルが見つかったかどうか) に応じて処理を変更
並列処理: 個別に実行できる複数のステップ (例: 場所、役割、小道具の分析)
合流ポイント: 複数の並列ステップが完了するのを待ってから続行

状態の管理

LangGraphを使用すると、ワークフロー全体で簡単に状態を管理できます。各ステップでは、現在の状態を受け取り、必要に応じて変更して、更新した状態を次のステップに渡します。

Pydanticによる構造化データ

Pydanticモデルを使用して、各ステップでデータの構造を定義します。これにより、型検証が行われ、ワークフローの堅牢性を維持できます。

AIエージェント用のBox APIツール

このワークフローでは、いくつかのBox APIツールを利用しています。

box_search_tool: 指定したクエリとフィルタを使用してBox内のファイルを検索する
box_read_tool: Box内のファイルのテキストコンテンツを読み取る
box_ask_ai_tool: Box AIを使用してファイルに関する質問に回答する
box_search_folder_by_name: Box内のフォルダを名前で検索する
box_ai_extract_data: Box AIを使用してファイルから構造化データを抽出する
box_list_folder_content_by_folder_id: Box内のフォルダのコンテンツのリストを取得する

これらのツールにより、AIエージェントは自然かつ体系的にBoxと対話できます。

まとめ

このワークショップでは、LangGraphとBox APIを使用してAIを活用したワークフローを構築しました。今回は、以下の方法について説明しました。

LangGraphのStateGraphを使用して、構造化されたワークフローを定義する
AIエージェント用のBox APIツールを統合する
順次実行、並列実行、条件付き実行などのさまざまなワークフローパターンを使用する
ワークフロー全体で状態を管理する
構造化データ用にPydanticモデルを作成する

今回紹介したのはほんの簡単な例です。このワークフローを拡張して、さまざまな種類のドキュメントを処理したり、分析ステップをさらに追加したり、人間によるレビューを組み込んだりすることができます。

次のステップ

このプロジェクトを拡張するためのアイデアをいくつか紹介します。

映画脚本以外の種類のドキュメントを追加する
他のAPIと統合して機能を追加する
分析結果の可視化機能を追加する
継続的な改善のためにフィードバックループを実装する
操作を容易にするためにウェブインターフェースを追加する

ぜひ試してみてください。