[PR #209] [MERGED] Detect nested workflows #318

New Issue

2026-02-16T01:16:52-05:00

yindo commented

2026-02-16 01:16:52 -05:00

📋 Pull Request Information

Original PR: https://github.com/run-llama/llama_deploy/pull/209
Author: @logan-markewich
Created: 8/29/2024
Status: ✅ Merged
Merged: 8/30/2024
Merged by: @logan-markewich

Base: v0.1.0 ← Head: logan/detect_nested_workflows

📝 Commits (4)

a0c73f3 add network service detection modules
3bea9f5 Merge branch 'v0.1.0' into logan/detect_nested_workflows
b2b1bfc update readme, fix small bugs
de5f7e6 fix missing queue args in tests

📊 Changes

4 files changed (+185 additions, -4 deletions)

View changed files

📝 README.md (+97 -0)
📝 llama_agents/deploy/deploy.py (+6 -0)
➕ llama_agents/deploy/network_workflow.py (+70 -0)
📝 tests/message_queues/test_simple_remote_client.py (+12 -4)

📄 Description

This PR adds the ability to detect and deploy nested workflows

It works by highjacking the ServiceManager that is inside each workflow. Before injecting the workflow into a step, it optionally checks if that workflow already exists as a service, and then swaps in a new workflow that just makes calls to the remote workflow.

This means users need zero code changes to their existing workflows, they just have to deploy everything, and reap the benefits :)

One concern I had was that the nested workflow always creates (and deletes) a new session. Tbh I wanted to reuse the existing session, but couldn't find a way to insert it. Maybe a future PR :)

Script1 (deploying core services)

from llama_agents import deploy_core, ControlPlaneConfig, SimpleMessageQueueConfig

async def main():
    await deploy_core(
        ControlPlaneConfig(),
        SimpleMessageQueueConfig(),
    )

if __name__ == "__main__":
    import asyncio

    loop = asyncio.get_event_loop()
    loop.run_until_complete(main())

Script2 (creating and deploying workflows)

import asyncio
from llama_index.core.workflow import Workflow, StartEvent, StopEvent, step

class InnerWorkflow(Workflow):

    @step()
    async def run_step(self, ev: StartEvent) -> StopEvent:
        arg1 = ev.get("arg1")
        if not arg1:
            raise ValueError("arg1 is required.")

        return StopEvent(result=str(arg1) + "_result")


class OuterWorkflow(Workflow):

    @step()
    async def run_step(self, ev: StartEvent, inner: InnerWorkflow) -> StopEvent:
        arg1 = ev.get("arg1")
        if not arg1:
            raise ValueError("arg1 is required.")
        
        arg1 = await inner.run(arg1=arg1)

        return StopEvent(result=str(arg1) + "_result")


inner = InnerWorkflow()
outer = OuterWorkflow()
outer.add_workflows(inner=InnerWorkflow())


from llama_agents import deploy_workflow, ControlPlaneConfig, WorkflowServiceConfig

async def main():
    inner_task = asyncio.create_task(
        deploy_workflow(
            inner,
            WorkflowServiceConfig(host="127.0.0.1", port=8003, service_name="inner"),
            ControlPlaneConfig(),
        )
    )

    outer_task = asyncio.create_task(
        deploy_workflow(
            outer,
            WorkflowServiceConfig(host="127.0.0.1", port=8002, service_name="outer"),
            ControlPlaneConfig(),
        )
    )

    await asyncio.gather(inner_task, outer_task)

if __name__ == "__main__":
    import asyncio
    
    loop = asyncio.get_event_loop()
    loop.run_until_complete(main())

Script3 (using the client)

from llama_agents import LlamaAgentsClient, ControlPlaneConfig

client = LlamaAgentsClient(ControlPlaneConfig())
session = client.create_session()
session.run("outer", arg1="hello_world")
> 'hello_world_result_result'

_{🔄 This issue represents a GitHub Pull Request. It cannot be merged through Gitea due to API limitations.}

## 📋 Pull Request Information **Original PR:** https://github.com/run-llama/llama_deploy/pull/209 **Author:** [@logan-markewich](https://github.com/logan-markewich) **Created:** 8/29/2024 **Status:** ✅ Merged **Merged:** 8/30/2024 **Merged by:** [@logan-markewich](https://github.com/logan-markewich) **Base:** `v0.1.0` ← **Head:** `logan/detect_nested_workflows` --- ### 📝 Commits (4) - [`a0c73f3`](https://github.com/run-llama/llama_deploy/commit/a0c73f30703e69ae5d1c76abb48f8130a4566180) add network service detection modules - [`3bea9f5`](https://github.com/run-llama/llama_deploy/commit/3bea9f5ac43591a175d8b4104b14d352154849ca) Merge branch 'v0.1.0' into logan/detect_nested_workflows - [`b2b1bfc`](https://github.com/run-llama/llama_deploy/commit/b2b1bfc0f9eb1c35651bb173a7418a4a72b3a40c) update readme, fix small bugs - [`de5f7e6`](https://github.com/run-llama/llama_deploy/commit/de5f7e67a8c440a3536fb4afc5a7a6eba80ca710) fix missing queue args in tests ### 📊 Changes **4 files changed** (+185 additions, -4 deletions) <details> <summary>View changed files</summary> 📝 `README.md` (+97 -0) 📝 `llama_agents/deploy/deploy.py` (+6 -0) ➕ `llama_agents/deploy/network_workflow.py` (+70 -0) 📝 `tests/message_queues/test_simple_remote_client.py` (+12 -4) </details> ### 📄 Description This PR adds the ability to detect and deploy nested workflows It works by highjacking the `ServiceManager` that is inside each workflow. Before injecting the workflow into a step, it optionally checks if that workflow already exists as a service, and then swaps in a new workflow that just makes calls to the remote workflow. This means users need zero code changes to their existing workflows, they just have to deploy everything, and reap the benefits :) --- One concern I had was that the nested workflow always creates (and deletes) a new session. Tbh I wanted to reuse the existing session, but couldn't find a way to insert it. Maybe a future PR :) --- Script1 (deploying core services) ```python from llama_agents import deploy_core, ControlPlaneConfig, SimpleMessageQueueConfig async def main(): await deploy_core( ControlPlaneConfig(), SimpleMessageQueueConfig(), ) if __name__ == "__main__": import asyncio loop = asyncio.get_event_loop() loop.run_until_complete(main()) ``` Script2 (creating and deploying workflows) ```python import asyncio from llama_index.core.workflow import Workflow, StartEvent, StopEvent, step class InnerWorkflow(Workflow): @step() async def run_step(self, ev: StartEvent) -> StopEvent: arg1 = ev.get("arg1") if not arg1: raise ValueError("arg1 is required.") return StopEvent(result=str(arg1) + "_result") class OuterWorkflow(Workflow): @step() async def run_step(self, ev: StartEvent, inner: InnerWorkflow) -> StopEvent: arg1 = ev.get("arg1") if not arg1: raise ValueError("arg1 is required.") arg1 = await inner.run(arg1=arg1) return StopEvent(result=str(arg1) + "_result") inner = InnerWorkflow() outer = OuterWorkflow() outer.add_workflows(inner=InnerWorkflow()) from llama_agents import deploy_workflow, ControlPlaneConfig, WorkflowServiceConfig async def main(): inner_task = asyncio.create_task( deploy_workflow( inner, WorkflowServiceConfig(host="127.0.0.1", port=8003, service_name="inner"), ControlPlaneConfig(), ) ) outer_task = asyncio.create_task( deploy_workflow( outer, WorkflowServiceConfig(host="127.0.0.1", port=8002, service_name="outer"), ControlPlaneConfig(), ) ) await asyncio.gather(inner_task, outer_task) if __name__ == "__main__": import asyncio loop = asyncio.get_event_loop() loop.run_until_complete(main()) ``` Script3 (using the client) ```python from llama_agents import LlamaAgentsClient, ControlPlaneConfig client = LlamaAgentsClient(ControlPlaneConfig()) session = client.create_session() session.run("outer", arg1="hello_world") > 'hello_world_result_result' ``` --- <sub>🔄 This issue represents a GitHub Pull Request. It cannot be merged through Gitea due to API limitations.</sub>

yindo added the pull-request label 2026-02-16 01:16:52 -05:00

yindo closed this issue

2026-02-16 01:16:52 -05:00

yindo referenced this issue

2026-02-16 01:17:10 -05:00

[PR #318] [MERGED] Update full stack example to have state-management #392

Sign in to join this conversation.

1 Participants

Notifications

Due Date

No due date set.

Dependencies

No dependencies set.

Reference: run-llama/llama_deploy#318