Parallelize

Parallelization takes many forms.

Simple

Declare agents in a for loop.

import parallem as pllm
from dotenv import load_dotenv


def power_of_n_agent(agt: pllm.AgentContext, n: int):
    return agt.ask_llm(f"Name a power of {n}").final_answer


load_dotenv()
with pllm.resume_directory(
    ".pllm/simplest",
    strategy="sync",  # or batch
) as orch:
    for i in range(2, 6):
        with orch.agent(i) as agt:
            print(power_of_n_agent(agt, i))

It is effective with sync and batch. However, it is not effective if ran asynchronously, because power-of-2 agent must finish before power-of-3 agent can begin.

Async

It is effective with sync, async, and batch strategies.

import asyncio
import random

import parallem as pllm


async def haiku_writer_agent(agt: pllm.AgentContext):
    # Declare the agent.
    conv = agt.get_msg_state()

    rand_time = random.uniform(1, 4)
    await asyncio.sleep(rand_time)  # Simulate some processing time
    await conv.ask_llm("Please name an animal in 1 word.")
    await conv.ask_llm(f"Write a haiku about {conv[-1].final_answer}(s).")
    out = conv[-1].final_answer
    print(out)
    return out


async def main():
    with pllm.resume_directory(
        ".pllm/simplest",
        provider="openai",
        strategy="batch",
        dashboard=True,
        load_dotenv=True,
        ask_params={
            "salt": 7,
        }
    ) as orch:
        coros = []
        for i in range(10):
            with orch.agent(f"Writer-{i}") as a:
                coros.append(haiku_writer_agent(a))

        await orch.gather(*coros)

if __name__ == "__main__":
    asyncio.run(main())

Note

In the above example, an alternative is asyncio.gather(*coros). But in batch mode, if the first child raises an error due to a missing value, that would prevent the remaining coroutines from being batched.

Therefore, either asyncio.gather(*coros, return_exceptions=True) or orch.gather(*coros) are recommended because it gracefully handles if a value is not available.

In the current version of ParaLLeM, strategy=async creates a separate event loop, and ask_llm eagerly kicks off that task. await conv.ask_llm ensures that that task is complete.