Skip to content

agentic_app.py

The agentic_app.py module is used in orchestrator-core for actually running the entire Agentic WFO FastAPI backend and the CLI.

FastAPI Backend

The code for the WFO's Fast API backend is very well documented, so look through the functions used in this module here:

orchestrator.agentic_app

The main application module.

This module contains the main LLMOrchestratorCore class for the FastAPI backend and provides the ability to run the CLI with LLM features (search and/or agent).

LLMOrchestratorCore

Bases: orchestrator.app.OrchestratorCore

Source code in orchestrator/agentic_app.py 
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
class LLMOrchestratorCore(OrchestratorCore):
    def __init__(
        self,
        *args: Any,
        llm_settings: LLMSettings = llm_settings,
        agent_model: "OpenAIChatModel | str | None" = None,
        **kwargs: Any,
    ) -> None:
        """Initialize the `LLMOrchestratorCore` class.

        This class extends `OrchestratorCore` with LLM features (search and agent).
        It runs the search migration based on feature flags.

        Args:
            *args: All the normal arguments passed to the `OrchestratorCore` class.
            llm_settings: A class of settings for the LLM
            agent_model: Override the agent model (defaults to llm_settings.AGENT_MODEL)
            **kwargs: Additional arguments passed to the `OrchestratorCore` class.

        Returns:
            None
        """
        self.llm_settings = llm_settings
        self.agent_model = agent_model or llm_settings.AGENT_MODEL

        super().__init__(*args, **kwargs)

        # Run search migration if search or agent is enabled
        if self.llm_settings.SEARCH_ENABLED or self.llm_settings.AGENT_ENABLED:
            logger.info("Running search migration")
            try:
                from orchestrator.db import db
                from orchestrator.search.llm_migration import run_migration

                with db.engine.begin() as connection:
                    run_migration(connection)
            except ImportError as e:
                logger.error(
                    "Unable to run search migration. Please install search dependencies: "
                    "`pip install orchestrator-core[search]`",
                    error=str(e),
                )
                raise
__init__ 
__init__(
    *args: typing.Any,
    llm_settings: orchestrator.llm_settings.LLMSettings = llm_settings,
    agent_model: pydantic_ai.models.openai.OpenAIChatModel | str | None = None,
    **kwargs: typing.Any
) -> None

Initialize the LLMOrchestratorCore class.

This class extends OrchestratorCore with LLM features (search and agent). It runs the search migration based on feature flags.

Parameters:

  • *args (typing.Any, default: () ) –

    All the normal arguments passed to the OrchestratorCore class.

  • llm_settings (orchestrator.llm_settings.LLMSettings, default: orchestrator.agentic_app.LLMOrchestratorCore.llm_settings ) –

    A class of settings for the LLM

  • agent_model (pydantic_ai.models.openai.OpenAIChatModel | str | None, default: None ) –

    Override the agent model (defaults to llm_settings.AGENT_MODEL)

  • **kwargs (typing.Any, default: {} ) –

    Additional arguments passed to the OrchestratorCore class.

Returns:

  • None –

    None

Source code in orchestrator/agentic_app.py 
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
def __init__(
    self,
    *args: Any,
    llm_settings: LLMSettings = llm_settings,
    agent_model: "OpenAIChatModel | str | None" = None,
    **kwargs: Any,
) -> None:
    """Initialize the `LLMOrchestratorCore` class.

    This class extends `OrchestratorCore` with LLM features (search and agent).
    It runs the search migration based on feature flags.

    Args:
        *args: All the normal arguments passed to the `OrchestratorCore` class.
        llm_settings: A class of settings for the LLM
        agent_model: Override the agent model (defaults to llm_settings.AGENT_MODEL)
        **kwargs: Additional arguments passed to the `OrchestratorCore` class.

    Returns:
        None
    """
    self.llm_settings = llm_settings
    self.agent_model = agent_model or llm_settings.AGENT_MODEL

    super().__init__(*args, **kwargs)

    # Run search migration if search or agent is enabled
    if self.llm_settings.SEARCH_ENABLED or self.llm_settings.AGENT_ENABLED:
        logger.info("Running search migration")
        try:
            from orchestrator.db import db
            from orchestrator.search.llm_migration import run_migration

            with db.engine.begin() as connection:
                run_migration(connection)
        except ImportError as e:
            logger.error(
                "Unable to run search migration. Please install search dependencies: "
                "`pip install orchestrator-core[search]`",
                error=str(e),
            )
            raise