How to Use Agent

To pipeline of how to use agents in AIOS is shown as the following figure.

AIOS Agent Integration Flow

Agent Download

AIOS will call AgentManager to download agents. The manager connects to a remote server to fetch the agent data.

# download agents from manager
def download_agent(self, author: str, name: str, version: str = "latest") -> tuple[str, str, str]:
    params = {
        "author": author,
        "name": name,
    }

    if version != 'latest':
        params['version'] = version
        cache_path = self._get_cache_path(author, name, version)
        if cache_path.exists():
            print(f"Using cached version of {author}/{name} (v{version})")
            return author, name, version
    else:
        cached_versions = sorted(
            self._get_cached_versions(author, name), reverse=True)
        if cached_versions:
            latest_cached = self._path_to_version(cached_versions[0])
            print(
                f"Using latest cached version of {author}/{name} (v{latest_cached})")
            return author, name, latest_cached

    response = requests.get(f"{self.base_url}/api/download", params=params)
    response.raise_for_status()
    agent_data = response.json()

    actual_version = agent_data.get('version', 'unknown')
    cache_path = self._get_cache_path(author, name, actual_version)

    self._save_agent_to_cache(agent_data, cache_path)
    print(
        f"Agent {author}/{name} (v{actual_version}) downloaded and cached successfully.")

    if not self.check_reqs_installed(cache_path):
        self.install_agent_reqs(cache_path)

    return author, name, actual_version

Agent Activation

AIOS will first check whether the agent already exists in the local environment. If the agent is unavailable locally, it will call the AgentManager to download agents. If both local and remote loading fails, it will raise an issue.

def activate_agent(self, agent_name: str, task_input):
    try:
        agent_class = self.load_agent_instance(agent_name)
    except:
        # If local loading fails, try downloading and loading
        try:
            agent_name = '/'.join(
                self.manager.download_agent(
                    *agent_name.split('/')
                ))
            agent_class = self.load_agent_instance(agent_name)
        except Exception as e:
            print(f"Warning: Both local and remote loading failed. Error: {str(e)}")
            raise

    agent = agent_class(
        agent_name=agent_name,
        task_input=task_input,
        log_mode=self.agent_log_mode
    )

    return agent

def load_agent_instance(self, compressed_name: str):
    # First try local loading
    try:
        author, name = compressed_name.split("/")
        module_name = ".".join(["pyopenagi", "agents", author, name, "agent"])
        class_name = self.snake_to_camel(name)
        agent_module = importlib.import_module(module_name)
        return getattr(agent_module, class_name)
    except (ImportError, AttributeError, ValueError):
        # If local loading fails, use original remote loading logic
        name_split = compressed_name.split('/')
        return self.manager.load_agent(*name_split)

Register and Submit Tasks

Register your agents and submit tasks for them to perform. Below is the example to submit and run agents.

with aios_starter(**vars(args)) as (submit_agent, await_agent_execution):
    # register your agents and submit agent tasks
    agent_tasks = [
        ["example/academic_agent", "Tell me what is the prollm paper mainly about? "],
    
    agent_ids = []
    for agent_name, task_input in agent_tasks:
        agent_id = submit_agent(agent_name=agent_name, task_input=task_input)
        agent_ids.append(agent_id)

    for agent_id in agent_ids:
        await_agent_execution(agent_id)