Repository: emcie-co/parlant
File: src/parlant/bin/server.py (Python)

Code before:
    return GeminiService(LOGGER)


def load_openai() -> NLPService:
    from parlant.adapters.nlp.openai import OpenAIService

    return OpenAIService(LOGGER)


def load_together() -> NLPService:
    from parlant.adapters.nlp.together import TogetherService

    return TogetherService(LOGGER)


async def create_agent_if_absent(agent_store: AgentStore) -> None:
    agents = await agent_store.list_agents()
    if not agents:
        await agent_store.create_agent(name=DEFAULT_AGENT_NAME)


@asynccontextmanager
async def load_modules(
    container: Container,
    modules: list[str],
) -> AsyncIterator[None]:
    if not modules:
        config_file = "parlant_config.toml"
        if os.path.isfile(config_file):
            config = toml.load(config_file)
            # Expecting structure of:
            # [parlant]
            # modules = ["module_1", "module_2"]
            modules = config.get("parlant", {}).get("modules", [])
        else:
            modules = []

    if not modules:
        container[Logger].info("no modules loaded")
        yield
        return

    imported_modules = []

    for module_path in modules:
        module = importlib.import_module(module_path)
        if not hasattr(module, "initialize_module") or not hasattr(module, "shutdown_module"):
            raise StartupError(
                f"Module {module.__name__} must define initialize_module and shutdown_module"
            )
        imported_modules.append(module)

Review comment (suggestion):
It's bad design to specifically provide a list of modules to this function, then go and do something else inside like read a file as a fallback.

This function should just get the exact modules that need to be run. And this function should **only** be run if the list of modules is non-empty.

Understanding which modules should be run based on config or params should be done in a separate function.

Code after:


def load_openai() -> NLPService:
    from parlant.adapters.nlp.openai import OpenAIService

    return OpenAIService(LOGGER)


def load_together() -> NLPService:
    from parlant.adapters.nlp.together import TogetherService

    return TogetherService(LOGGER)


async def create_agent_if_absent(agent_store: AgentStore) -> None:
    agents = await agent_store.list_agents()
    if not agents:
        await agent_store.create_agent(name=DEFAULT_AGENT_NAME)


async def get_module_list_from_config() -> list[str]:
    config_file = Path("parlant.toml")

    if config_file.exists():
        config = toml.load(config_file)
        # Expecting structure of:
        # [parlant]
        # modules = ["module_1", "module_2"]
        return list(config.get("parlant", {}).get("modules", []))

    return []


@asynccontextmanager
async def load_modules(
    container: Container,
    modules: Iterable[str],
) -> AsyncIterator[None]:
    imported_modules = []

    for module_path in modules:
        module = importlib.import_module(module_path)
        if not hasattr(module, "initialize_module") or not hasattr(module, "shutdown_module"):
            raise StartupError(
                f"Module '{module.__name__}' must define initialize_module(container: lagom.Container) and shutdown_module()"
            )
        imported_modules.append(module)

    for m in imported_modules:
        LOGGER.info(f"Initializing module '{m.__name__}'")
        await m.initialize_module(container)