Building AI-Powered workflows with LangGraph and Box API

By combining the power of Large Language Models (LLMs) with structured workflows and cloud storage APIs, we can create intelligent systems that can analyze, process, and generate insights from documents stored in the cloud.

In this workshop, we’ll explore how to build AI-powered workflows using LangGraph and the Box API. We’ll create a sample application that processes movie scripts stored in Box, analyzes their content, and generates comprehensive reports with details about characters, locations, props, and more.

Here is a video about this example:

What you’ll learn

How to set up a LangGraph workflow for document processing
How to integrate Box API with AI agents
Different workflow patterns (sequential, parallel, conditional)
How to use Box API tools for AI agents
How to structure and organize state in your workflow
Best practices for building robust AI-powered workflows

Understanding LangGraph

LangGraph is a library for building stateful, multi-actor applications with LLMs. It allows you to:

Define structured workflows with different states and transitions
Maintain state between steps in your workflow
Create complex workflows with conditional branching, parallel execution, and more
Integrate with various tools and APIs

The core concept in LangGraph is the StateGraph, which represents your workflow as a directed graph where:

Nodes are processing steps (functions that transform the state)
Edges define the flow between steps
State is passed from one step to the next

Box API integration

The Box API allows us to interact with Box content cloud, enabling our AI workflow to:

Search for files and folders
Read file content
Use Box AI features like data extraction and answering questions
List folder contents
Create, update, and delete files and folders

We’ll use the box-ai-agents-toolkit package to simplify our Box API integration.

Designing the workflow

Let’s design a workflow that processes movie scripts stored in Box:

Locate a script file in Box
Read the script content
Analyze the script for key information
Locations mentioned in the script
Character roles in the script
Props used in the script
Suggest actors for each character role
Analyze the script’s author
Suggest suitable producers and directors
Generate a comprehensive report

Here’s a visual representation of our workflow:

Defining the data models

First, let’s define Pydantic models to represent our data at each step of the workflow. These models provide structure and type validation for our workflow state.

# 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

Setting up Box API tools

Next, let’s set up our Box API tools that our AI agents will use:

# 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
# ...

Implementing the workflow steps

Now, let’s implement the individual workflow steps as functions that transform our workflow state:

# 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
# ...

Building the workflow graph

Now, let’s build our workflow graph by defining the nodes (steps) and edges (transitions):

# 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()

Running the workflow

Finally, let’s create a script to run our workflow:

# 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()

Key LangGraph features

Let’s highlight the key LangGraph features we’ve demonstrated:

StateGraph for workflow definition

LangGraph’s StateGraph allows us to define our workflow as a directed graph with nodes and edges. This provides a clear, visual representation of our workflow and makes it easy to reason about the flow of data.

Workflow patterns

We’ve demonstrated several important workflow patterns:

Sequential processing: Steps that follow one after another (e.g., locate file → read file → analyze script)
Conditional branching: Different paths based on conditions (e.g., whether the file was found)
Parallel processing: Multiple steps that can run independently (e.g., analyzing locations, roles, and props)
Join points: Waiting for multiple parallel steps to complete before continuing

State management

LangGraph makes it easy to manage state throughout the workflow. Each step receives the current state, modifies it as needed, and passes the updated state to the next step(s).

Structured data with Pydantic

We use Pydantic models to define the structure of our data at each step. This provides type validation and helps ensure that our workflow remains robust.

Box API tools for AI agents

Our workflow leverages several Box API tools:

box_search_tool: Searches for files in Box using specified queries and filters
box_read_tool: Reads the text content of a file in Box
box_ask_ai_tool: Uses Box AI to answer questions about a file
box_search_folder_by_name: Locates a folder in Box by its name
box_ai_extract_data: Extracts structured data from a file using Box AI
box_list_folder_content_by_folder_id: Lists the content of a folder in Box

These tools allow our AI agents to interact with Box in a natural, structured way.

Conclusion

In this workshop, we’ve built an AI-powered workflow using LangGraph and Box API. We’ve learned how to:

Define a structured workflow using LangGraph’s StateGraph
Integrate Box API tools for AI agents
Use different workflow patterns like sequential, parallel, and conditional execution
Manage state throughout the workflow
Create Pydantic models for structured data

This example is just the beginning. You can extend this workflow to handle different types of documents, add more analysis steps, incorporate human review, and much more.

Next Steps

Here are some ideas for extending this project:

Add more document types beyond movie scripts
Integrate with other APIs for additional functionality
Add visualization capabilities for the analysis results
Implement feedback loops for continuous improvement
Add a web interface for easier interaction

Happy coding!