Building Dynamic Agentic Workflows at Runtime

[McCReuben]

Hello LangGraph community!

I've been exploring the capabilities of LangGraph and wanted to share an interesting implementation I've been working on. This project is a part of an LLM e-book reading assistant called Pyri that me and a partner have been working on for months. It focuses on by designing, building, and compiling graphs at runtime based on a user query.

Overview

The implementation is inspired by the LLM Compiler implementation from a few months back, but executes DAG plans via Langgraph nodes instead of tools, while maintaining both sequential and parallel execution capabilities. It consists of two main stages:

1. Planning Stage: A high-level plan DAG is generated based on the user's query. This plan outlines the necessary steps and selects appropriate agents to fulfill the request.
2. Execution Stage: The generated plan is converted into a LangGraph StateGraph object. The graph is then executed, orchestrating multiple agents that run sequentially or in parallel, subject to their dependencies.

Notable Features

• Runtime Graph Design and Compilation: Instead of pre-defining workflows, the graphs are dynamically created at runtime via a Planner agent. This allows the system to adapt to any query by generating a tailored execution plan.
• Node execution: By executing plan steps as nodes instead of tools (like in LLM Compiler), we get better insight tools via LangSmith and functionalities like Human-in-the-loop.
• Support for Multiple Agents: The system can orchestrate numerous agents, executing them sequentially or in parallel based on the plan's dependencies. This enhances efficiency and scalability.
• Scalable Graph Complexity: The complexity of the execution graph scales with the complexity of the user's query. More complex queries result in more intricate plans and graphs.

Code Breakdown

Planning Agent

The planning agent uses a language model (e.g., GPT-4) to generate a structured plan. It considers the user's context, available agents, and conversation history to create a detailed plan in JSON format.

class PlannerState(BaseModel):
    user_chat_message_and_context: UserChatMessageAndContext
    llm_model: str = "gpt-4o"
    selected_plan: Optional[Plan] = None
    subgraph_result: Optional[str] = None
    final_answer: Optional[str] = None

def create_plan(state: PlannerState) -> PlannerState:
    llm = ChatOpenAI(model=state.llm_model)
    structured_llm = llm.with_structured_output(Plan)
    chain = planner_prompt | structured_llm
    # Prepare context and messages
    plan = chain.invoke({
        "messages": messages,
        "book_context": book_context_str,
        "user_context": user_str,
        "agent_descriptions": "\n---\n".join(agent_descriptions),
    })
    return PlannerState(
        user_chat_message_and_context=state.user_chat_message_and_context,
        selected_plan=plan,
    )


def finalize_answer(state: PlannerState) -> PlannerState:
    ...


async def execute_subgraph(state: PlannerState) -> PlannerState:
    if state.selected_plan is None:
        raise ValueError("No plan selected")

    plan_graph = plan_to_graph(state.selected_plan)
    step_data = initialize_step_data(state.selected_plan)
    initial_subgraph_state = SubPlanState(
        user_chat_message_and_context=state.user_chat_message_and_context.model_dump(),
        step_data=step_data,
    )

def create_planner_graph():
    planner_graph = StateGraph(PlannerState)

    planner_graph.add_node("create_plan", create_plan)
    planner_graph.add_node("execute_subgraph", execute_subgraph)
    planner_graph.add_node("finalize_answer", finalize_answer)

    planner_graph.add_edge(START, "create_plan")
    planner_graph.add_edge("create_plan", "execute_subgraph")
    planner_graph.add_edge("execute_subgraph", "finalize_answer")
    planner_graph.add_edge("finalize_answer", END)

    return planner_graph.compile()

Execution of the Plan as a Graph

The generated plan is transformed into a StateGraph object. Each step in the plan becomes a node in the graph, and dependencies define the edges between nodes.

State and types:

class Step(BaseModel):
    id: str = Field(
        description="The step number, an integer starting from 1 and incrementing by 1 for each step"
    )
    action: str = Field(description="The action to be taken in the step")
    depends_on: List[str] = Field(
        description="A list of step ids that must be completed before this step can be executed"
    )
    args: Optional[Dict[str, Any]] = Field(
        description="A dictionary representing the arguments to be passed to the action, with dependent inputs being passed as $id"
    )
    description: Optional[str] = Field(
        description="A description of what the step accomplishes and how it fits into the plan"
    )

class Plan(BaseModel):
    steps: List[Step] = Field(description="A list of steps in the plan")
    description: Optional[str] = Field(description="A description of the plan")

class StepResult(TypedDict):
    input: Dict[str, Any]
    result: Optional[str] = None

class SubPlanState(BaseModel):
    user_chat_message_and_context: UserChatMessageAndContext
    plan: Optional[Plan] = None
    step_data: Dict[str, Step]
    agent_results: Annotated[Dict[str, StepResult], dict_concat] = {}
    final_answer: Optional[str] = None

Helper functions:

def substitute_variables(
    args: Dict[str, Any], depends_on: List[str], agent_results: Dict[str, Step]
) -> Dict[str, Any]:
    """
    Substitutes input parameters when they are dependent on the results of previous steps
    Input:
        args: A dictionary of arguments, with dependent step id outputs in arg values being denoted by $id
        depends_on: The list of step ids that the inputs depend on
        agent_results: A dictionary of agent results, where the keys are step ids and the values are dictionaries containing the input and result of the step
    Output:
        A dictionary of arguments, with variables substituted by the results from the agent_results dictionary
    """
    updated_args = {}
    for arg_key, arg in args.items():
        pattern = r"\$(\d+)"
        while re.search(pattern, arg):
            arg = re.sub(
                pattern,
                lambda m: agent_results.get(str(m.group(1)), {}).get("result", ""),
                arg,
            )
        updated_args[arg_key] = arg
    return updated_args


def generic_node_function(state: SubPlanState, step_id: str, method_to_call):
    """
    Wrapper for every step / agent in the plan.
    Input:
        state: The current state of the plan
        step_id: The id of the step to execute (step number)
        method_to_call: The method to call to execute the step
    Output:
        The result of the step
    """
    logger.info(f"Executing step #{step_id}")
    step_info = state.step_data[step_id]
    args = step_info.args
    depends_on = step_info.depends_on
    step_updated_args = args.copy()
    # Substitute variables with results from previous steps
    if depends_on:
        step_updated_args = substitute_variables(
            step_updated_args, depends_on, state.agent_results
        )
    try:
        # Call the method with updated args
        if asyncio.iscoroutinefunction(method_to_call):
            result = method_to_call(state, **step_updated_args)
            # result = await method_to_call(state, **step_updated_args)
            logger.info(f"Async method {method_to_call} returned: {result}")
        else:
            result = method_to_call(state, **step_updated_args)
            # logger.info(f"Sync method {method_to_call} returned: {result}")
        is_final_step = int(step_id) == len(state.step_data)
        agent_results = {step_id: {"input": step_updated_args, "result": result}}
        state_updated = {"agent_results": agent_results}
        if is_final_step:
            state_updated["final_answer"] = result
        return state_updated

    except Exception as e:
        logger.info(f"Error executing step #{step_id}: {e}")
        raise

Translating a Plan to a graph:

def plan_to_graph(plan: Plan):
    state_graph = StateGraph(SubPlanState)
    action_methods = {
        "ask_user_clarifying_question": ask_user_clarifying_question_method,
        "web_search": web_search_method,
        # ... (other actions / agents)
    }
    # Add nodes
    for step in plan.steps:
        step_id = step.id
        action = step.action
        node_name = f"{action}_{step_id}"
        method_to_call = action_methods[action]
        node_func = create_action_node(step_id, method_to_call)
        state_graph.add_node(node_name, node_func)
    # Add edges based on dependencies
    for step in plan.steps:
        node_name = f"{step.action}_{step.id}"
        if step.depends_on:
            for dep_id in step.depends_on:
                dep_node_name = f"{plan.step_data[dep_id].action}_{dep_id}"
                state_graph.add_edge(dep_node_name, node_name)
        else:
            state_graph.add_edge(START, node_name)
    # Add edges to END node
    # ... (additional code)
    compiled_graph = state_graph.compile()
    return compiled_graph

I am currently working on implementing Asyncronous execution and dynamic Human-in-the-loop capabilities. I would be happy to share a more complete version of the code with prompts and sub agents if there is interest. I'm loving the capabilities that LangGraph and LangSmith provide!

Happy coding!

[ajinkyaT]

@McCReuben That's quite an interesting use case. We are building something similar for our use case but with dynamic Human-in-the-loop capabilities for sub-graph/tool nodes, to correct the path taken and have human approval for sensitive tools. Please share your implementation for human in the loop if you are able to figure it out. Thanks!