academy.run

academy/run.py

HttpExchangeModel pydantic-model

Bases: BaseModel

Model for HttpExchangeFactory argument validation.

Config:

• extra: forbid

Fields:

• exchange_type (Literal['http'])
• url (str)
• auth_method (Literal['globus'] | None)
• additional_headers (dict[str, str] | None)
• request_timeout_s (float)
• ssl_verify (bool | None)

HybridExchangeModel pydantic-model

Bases: BaseModel

Model for HybridExchangeFactory argument validation.

Config:

• extra: forbid

Fields:

• exchange_type (Literal['hybrid'])
• redis_host (str)
• redis_port (int)
• redis_kwargs (dict[str, str])
• interface (str | None)
• namespace (str)
• ports (list[int] | None)

RedisExchangeModel pydantic-model

Bases: BaseModel

Model for RedisExchangeFactory argument validation.

Config:

• extra: allow

Fields:

• exchange_type (Literal['redis'])
• hostname (str)
• port (int)

GlobusExchangeModel pydantic-model

Bases: BaseModel

Model for GlobusExchangeFactory argument validation.

Config:

• extra: forbid

Fields:

• exchange_type (Literal['globus'])
• project_id (UUID)
• client_params (dict[str, Any])
• request_timeout_s (float)

AgentModel pydantic-model

Bases: BaseModel

Model to construct agent from yaml or pickle.

Config:

• extra: allow

Fields:

• constructor (str | None)
• pickle (Path | None)

Validators:

• check_constructor_or_pickle

check_constructor_or_pickle pydantic-validator

check_constructor_or_pickle() -> Self

Ensures either the constructor or pickle file is non null.

Source code in academy/run.py

@model_validator(mode='after')
def check_constructor_or_pickle(self) -> Self:
    """Ensures either the constructor or pickle file is non null."""
    if self.constructor is None and self.pickle is None:
        raise ValueError(
            'Either agent constructor or pickle file must be provided',
        )
    return self

AgentProcessConfig pydantic-model

Bases: BaseModel, Generic[AgentT]

Config for running an agent as an independent process.

Config:

• extra: forbid

Fields:

• exchange (HttpExchangeModel | HybridExchangeModel | RedisExchangeModel | GlobusExchangeModel)
• agent_registration (HttpAgentRegistration[AgentT] | HybridAgentRegistration[AgentT] | RedisAgentRegistration[AgentT] | GlobusAgentRegistration[AgentT] | None)
• agent (AgentModel)
• config (RuntimeConfig)

get_exchange

get_exchange() -> ExchangeFactory[Any]

Get the exchange factory specified by the config.

Source code in academy/run.py

def get_exchange(self) -> ExchangeFactory[Any]:
    """Get the exchange factory specified by the config."""
    if self.exchange.exchange_type == 'http':
        return HttpExchangeFactory(
            **self.exchange.model_dump(exclude={'exchange_type'}),
        )
    elif self.exchange.exchange_type == 'hybrid':
        return HybridExchangeFactory(
            **self.exchange.model_dump(exclude={'exchange_type'}),
        )
    elif self.exchange.exchange_type == 'redis':
        return RedisExchangeFactory(
            **self.exchange.model_dump(exclude={'exchange_type'}),
        )
    elif self.exchange.exchange_type == 'globus':
        return GlobusExchangeFactory(
            **self.exchange.model_dump(exclude={'exchange_type'}),
        )

    raise AssertionError('Unreachable')

get_agent

get_agent() -> AgentT

Get the agent specified by the config.

Source code in academy/run.py

def get_agent(self) -> AgentT:
    """Get the agent specified by the config."""
    agent: AgentT
    if self.agent.pickle:
        with open(self.agent.pickle, 'rb') as fp:
            agent = pickle.load(fp)
        if not isinstance(agent, Agent):
            raise TypeError(
                'Pickled class is not of type Agent.',
            )
        return agent
    elif self.agent.constructor:
        module_path, _, name = self.agent.constructor.rpartition('.')
        if len(module_path) == 0:
            raise ValueError(
                'Agent constructor must be the fully qualified path of '
                f'an agent constructor. Got "{self.agent.constructor}".',
            )
        module = importlib.import_module(module_path)
        agent_const = getattr(module, name)
        agent = agent_const(**self.agent.model_extra)  # type: ignore[arg-type]
        if not isinstance(agent, Agent):
            raise TypeError(
                'Agent constructor did not return an Agent type.',
            )
        return agent

    raise AssertionError('Unreachable')

to_toml

to_toml(filepath: str | Path) -> None

Write the config to a toml file.

Source code in academy/run.py

def to_toml(self, filepath: str | pathlib.Path) -> None:
    """Write the config to a toml file."""
    with open(filepath, 'wb') as f:
        tomli_w.dump(self.model_dump(exclude_none=True, mode='json'), f)

load classmethod

load(filepath: str | Path) -> Self

Parse an TOML config file.

Source code in academy/run.py

@classmethod
def load(cls, filepath: str | pathlib.Path) -> Self:
    """Parse an TOML config file."""
    with open(filepath, 'rb') as f:
        data = tomllib.load(f)
    return cls.model_validate(data)