Python Parallel AI Agents

example that illustrates how you might evolve the simple iterative planner–critic loop into a parallel agents design. In this example, multiple planner agents generate their plans concurrently, and a critic agent evaluates them in parallel. Then an aggregation step selects the best plan (or signals that further iteration is needed), which is similar to the approach used by systems like Convergence AI.

This example uses Python's asyncio to run the planner and critic calls concurrently. Note that in a production system you might add more sophisticated aggregation, iterative refinement, and real-time inter-agent communication.

import asyncio

from langchain.llms import OpenAI

# Initialize the LLM (assuming an async call method is available, e.g., acall)

llm = OpenAI(temperature=0.3)

# Custom prompts for planner and critic agents.

planner_prompt = """

You are an autonomous planning agent. Given the following goal:

{goal}

Develop a detailed, step-by-step plan to achieve this goal.

Return only the plan.

"""

critic_prompt = """

You are a critical evaluator. Assess the following plan for achieving the goal:

Goal: {goal}

Plan:

{plan}

If the plan is robust, complete, and the steps are clear, reply with "Approved".

Otherwise, provide specific feedback and suggestions to improve the plan.

"""

async def get_plan_async(goal, feedback=None):

"""

Asynchronously generate a plan from a planner agent.

Optionally include feedback to refine the plan.

"""

additional_context = f"\nHere is the previous feedback: {feedback}" if feedback else ""

prompt = planner_prompt.format(goal=goal) + additional_context

plan = await llm.acall(prompt)

return plan.strip()

async def critique_plan_async(goal, plan):

"""

Asynchronously have a critic agent evaluate a given plan.

"""

prompt = critic_prompt.format(goal=goal, plan=plan)

critique = await llm.acall(prompt)

return critique.strip()

async def parallel_planner_critic(goal, num_planners=3, max_iterations=3):

"""

Run multiple planner agents in parallel to generate plans, then evaluate them in parallel with a critic.

Repeat iteratively until one plan is approved or max iterations are reached.

"""

feedback = None

for iteration in range(1, max_iterations + 1):

print(f"\n=== Iteration {iteration} ===")

# Launch multiple planner agents concurrently.

planner_tasks = [asyncio.create_task(get_plan_async(goal, feedback)) for _ in range(num_planners)]

plans = await asyncio.gather(*planner_tasks)

print("Generated Plans:")

for i, plan in enumerate(plans, 1):

print(f"Plan {i}: {plan}\n")

# Have a critic evaluate each plan concurrently.

critic_tasks = [asyncio.create_task(critique_plan_async(goal, plan)) for plan in plans]

critiques = await asyncio.gather(*critic_tasks)

print("Critic Feedbacks:")

for i, critique in enumerate(critiques, 1):

print(f"Critique for Plan {i}: {critique}\n")

# Check if any plan is approved.

approved_plans = [plan for plan, critique in zip(plans, critiques) if "Approved" in critique]

if approved_plans:

best_plan = approved_plans[0] # Could use additional criteria to select among approved plans.

print("Selected Approved Plan:")

print(best_plan)

return best_plan

# If no plan is approved, aggregate feedback for further refinement.

aggregated_feedback = " | ".join(critiques)

feedback = aggregated_feedback # Feed aggregated feedback back into the planners.

print("No plan approved. Aggregated feedback for next iteration:")

print(feedback)

print("Max iterations reached. Using the last generated plan.")

return plans[0] # Fallback to the first plan if none was approved.

# Define the goal for our multi-agent system.

goal = (

"List all files in the current directory and then write and execute a Python function "

"that calculates the sum of 15 and 27."

)

# Run the parallel planner–critic process.

if __name__ == "__main__":

best_plan = asyncio.run(parallel_planner_critic(goal))

print("\nFinal Approved or Fallback Plan:\n", best_plan)