run_cli.py

# Import required modules and components
import os
os.environ["OMAGENT_MODE"] = "lite"
from pathlib import Path

from agent.conclude.conclude import Conclude
from agent.video_preprocessor.video_preprocess import VideoPreprocessor
from agent.video_qa.qa import VideoQA
from omagent_core.advanced_components.workflow.dnc.workflow import DnCWorkflow
from omagent_core.clients.devices.cli import DefaultClient
from omagent_core.engine.automator.task_handler import TaskHandler
from omagent_core.engine.workflow.conductor_workflow import ConductorWorkflow
from omagent_core.engine.workflow.task.do_while_task import (DnCLoopTask,
                                                             InfiniteLoopTask)
from omagent_core.engine.workflow.task.set_variable_task import SetVariableTask
from omagent_core.engine.workflow.task.simple_task import simple_task
from omagent_core.engine.workflow.task.switch_task import SwitchTask
from omagent_core.utils.build import build_from_file
from omagent_core.utils.container import container
from omagent_core.utils.logger import logging
from omagent_core.utils.registry import registry

logging.init_logger("omagent", "omagent", level="INFO")

# Set current working directory path
CURRENT_PATH = root_path = Path(__file__).parents[0]

# Import registered modules
registry.import_module(project_path=CURRENT_PATH.joinpath("agent"))

# Load container configuration from YAML file
container.register_stm("SharedMemSTM")
container.register_ltm(ltm="VideoMilvusLTM")
container.from_config(CURRENT_PATH.joinpath("container.yaml"))

# Initialize simple VQA workflow
workflow = ConductorWorkflow(name="video_understanding")

# 1. Video preprocess task for video preprocessing
video_preprocess_task = simple_task(
    task_def_name=VideoPreprocessor, task_reference_name="video_preprocess"
)

# 2. Video QA task for video QA
video_qa_task = simple_task(
    task_def_name=VideoQA,
    task_reference_name="video_qa",
    inputs={
        "video_md5": video_preprocess_task.output("video_md5"),
        "video_path": video_preprocess_task.output("video_path"),
        "instance_id": video_preprocess_task.output("instance_id"),
    },
)

dnc_workflow = DnCWorkflow()
dnc_workflow.set_input(query=video_qa_task.output("query"))

# 7. Conclude task for task conclusion
conclude_task = simple_task(
    task_def_name=Conclude,
    task_reference_name="task_conclude",
    inputs={
        "dnc_structure": dnc_workflow.dnc_structure,
        "last_output": dnc_workflow.last_output,
    },
)


# Configure workflow execution flow: Input -> Initialize global variables -> DnC Loop -> Conclude
workflow >> video_preprocess_task >> video_qa_task >> dnc_workflow >> conclude_task

# Register workflow
workflow.register(overwrite=True)

# Initialize and start app client with workflow configuration
cli_client = DefaultClient(
    interactor=workflow, config_path="configs"
)
cli_client.start_interactor()