LLM Multi-Agent Architecture in Python

Introduction to LLM Multi-Agent Architecture

Large Language Models (LLMs) have revolutionized natural language processing. Multi-agent architectures leverage multiple LLMs to solve complex tasks collaboratively. This approach enhances problem-solving capabilities and creates more robust AI systems.

import openai

def create_agents(num_agents):
    agents = []
    for i in range(num_agents):
        agent = openai.ChatCompletion.create(
            model="gpt-3.5-turbo",
            messages=[{"role": "system", "content": f"You are Agent {i+1}."}]
        )
        agents.append(agent)
    return agents

# Create 3 agents
agents = create_agents(3)
print(f"Created {len(agents)} agents")

Core Components of LLM Multi-Agent Systems

Multi-agent systems consist of several key components: agents (individual LLMs), a communication protocol, a task allocator, and a result aggregator. These components work together to distribute tasks, share information, and combine individual outputs into a cohesive solution.

class MultiAgentSystem:
    def __init__(self, agents):
        self.agents = agents
        self.task_queue = []
        self.results = []

    def add_task(self, task):
        self.task_queue.append(task)

    def allocate_tasks(self):
        for i, task in enumerate(self.task_queue):
            agent = self.agents[i % len(self.agents)]
            result = agent.process_task(task)
            self.results.append(result)

    def aggregate_results(self):
        return " ".join(self.results)

# Usage
mas = MultiAgentSystem(agents)
mas.add_task("Analyze market trends")
mas.add_task("Generate product ideas")
mas.allocate_tasks()
final_result = mas.aggregate_results()
print(f"Final result: {final_result}")

Agent Communication Protocols

Effective communication between agents is crucial for collaborative problem-solving. Protocols define how agents exchange information, request assistance, and share partial results. Common protocols include message passing, blackboard systems, and publish-subscribe models.

import asyncio

class Agent:
    def __init__(self, name):
        self.name = name
        self.inbox = asyncio.Queue()

    async def send_message(self, recipient, message):
        await recipient.inbox.put((self.name, message))

    async def receive_message(self):
        sender, message = await self.inbox.get()
        print(f"{self.name} received from {sender}: {message}")

async def simulate_communication():
    agent1 = Agent("Agent1")
    agent2 = Agent("Agent2")

    await agent1.send_message(agent2, "Hello, can you help with task X?")
    await agent2.receive_message()
    await agent2.send_message(agent1, "Sure, I can assist with task X.")
    await agent1.receive_message()

asyncio.run(simulate_communication())

Task Decomposition and Allocation

Complex tasks are broken down into smaller, manageable subtasks. These subtasks are then allocated to different agents based on their capabilities and current workload. This process optimizes resource utilization and enables parallel processing.

import random

class TaskManager:
    def __init__(self, agents):
        self.agents = agents
        self.tasks = []

    def add_task(self, task, complexity):
        self.tasks.append((task, complexity))

    def decompose_task(self, task, complexity):
        subtasks = []
        while complexity > 0:
            subtask_complexity = min(random.randint(1, 3), complexity)
            subtasks.append((f"{task} - Part {len(subtasks)+1}", subtask_complexity))
            complexity -= subtask_complexity
        return subtasks

    def allocate_tasks(self):
        for task, complexity in self.tasks:
            subtasks = self.decompose_task(task, complexity)
            for subtask, subtask_complexity in subtasks:
                agent = random.choice(self.agents)
                print(f"Assigning '{subtask}' (complexity: {subtask_complexity}) to {agent}")

# Usage
agents = ["Agent1", "Agent2", "Agent3"]
manager = TaskManager(agents)
manager.add_task("Develop marketing strategy", 5)
manager.add_task("Analyze competitor products", 3)
manager.allocate_tasks()

Consensus and Conflict Resolution

When multiple agents work on related tasks, they may produce conflicting results. Consensus mechanisms help resolve these conflicts and ensure a coherent final output. Common approaches include voting, weighted averaging, and hierarchical decision-making.

import numpy as np

class ConsensusResolver:
    def __init__(self, agents):
        self.agents = agents

    def resolve(self, opinions):
        if all(isinstance(o, (int, float)) for o in opinions):
            return np.mean(opinions)
        elif all(isinstance(o, str) for o in opinions):
            return max(set(opinions), key=opinions.count)
        else:
            raise ValueError("Unsupported opinion type")

def get_agent_opinions(agents, question):
    opinions = []
    for agent in agents:
        opinion = agent.get_opinion(question)
        opinions.append(opinion)
        print(f"{agent.name}: {opinion}")
    return opinions

# Simulating agents and opinions
agents = [Agent(f"Agent{i}") for i in range(1, 4)]
question = "What's the best approach for this problem?"
opinions = get_agent_opinions(agents, question)

resolver = ConsensusResolver(agents)
consensus = resolver.resolve(opinions)
print(f"Consensus: {consensus}")

Learning and Adaptation in Multi-Agent Systems

Multi-agent systems can improve their performance over time through learning and adaptation. Agents can share knowledge, learn from each other's successes and failures, and dynamically adjust their strategies based on the overall system performance.

import random

class AdaptiveAgent:
    def __init__(self, name, learning_rate=0.1):
        self.name = name
        self.knowledge = {}
        self.learning_rate = learning_rate

    def solve_problem(self, problem):
        if problem in self.knowledge:
            return self.knowledge[problem]
        else:
            return random.choice(["A", "B", "C"])

    def learn(self, problem, solution, success):
        if success:
            self.knowledge[problem] = solution
        elif problem in self.knowledge:
            if random.random() < self.learning_rate:
                del self.knowledge[problem]

def simulate_adaptive_system(agents, problems, iterations):
    for _ in range(iterations):
        problem = random.choice(problems)
        agent = random.choice(agents)
        solution = agent.solve_problem(problem)
        success = (solution == "A")  # Assume "A" is always the correct solution
        agent.learn(problem, solution, success)
        print(f"{agent.name} solved {problem} with {solution}. Success: {success}")

# Usage
agents = [AdaptiveAgent(f"Agent{i}") for i in range(1, 4)]
problems = ["P1", "P2", "P3"]
simulate_adaptive_system(agents, problems, 10)

Emergent Behavior in Multi-Agent Systems

Multi-agent systems often exhibit emergent behavior, where the collective actions of individual agents lead to complex, system-level patterns or solutions that weren't explicitly programmed. This phenomenon can result in innovative problem-solving approaches and unexpected insights.

import random
import matplotlib.pyplot as plt

class ForagingAgent:
    def __init__(self, x, y):
        self.x = x
        self.y = y

    def move(self):
        self.x += random.choice([-1, 0, 1])
        self.y += random.choice([-1, 0, 1])

def simulate_foraging(num_agents, iterations):
    agents = [ForagingAgent(0, 0) for _ in range(num_agents)]
    food_source = (10, 10)
    found_food = []

    for _ in range(iterations):
        for agent in agents:
            agent.move()
            if (agent.x, agent.y) == food_source:
                found_food.append(_ + 1)

    plt.hist(found_food, bins=20)
    plt.title("Distribution of Food Discovery Times")
    plt.xlabel("Iteration")
    plt.ylabel("Number of Agents")
    plt.show()

# Run simulation
simulate_foraging(num_agents=100, iterations=1000)

Real-life Example: Collaborative Writing Assistant

A multi-agent LLM system can be used to create a collaborative writing assistant. Different agents specialize in various aspects of writing, such as ideation, structure, grammar, and style. They work together to help users create high-quality content.

class WritingAssistant:
    def __init__(self):
        self.agents = {
            'ideation': Agent("Ideation Specialist"),
            'structure': Agent("Structure Expert"),
            'grammar': Agent("Grammar Checker"),
            'style': Agent("Style Consultant")
        }

    def generate_content(self, topic):
        ideas = self.agents['ideation'].generate_ideas(topic)
        outline = self.agents['structure'].create_outline(ideas)
        draft = self.agents['grammar'].write_draft(outline)
        final_content = self.agents['style'].refine_content(draft)
        return final_content

# Usage
assistant = WritingAssistant()
content = assistant.generate_content("The impact of AI on education")
print(content)

Real-life Example: Multi-Agent Customer Support System

A multi-agent LLM system can enhance customer support by routing queries to specialized agents, collaborating on complex issues, and providing consistent, high-quality responses across various channels.

import random

class CustomerSupportAgent:
    def __init__(self, name, specialization):
        self.name = name
        self.specialization = specialization

    def handle_query(self, query):
        return f"{self.name} ({self.specialization}): Responding to '{query}'"

class CustomerSupportSystem:
    def __init__(self):
        self.agents = [
            CustomerSupportAgent("Agent1", "General Inquiries"),
            CustomerSupportAgent("Agent2", "Technical Support"),
            CustomerSupportAgent("Agent3", "Billing"),
            CustomerSupportAgent("Agent4", "Product Information")
        ]

    def route_query(self, query):
        agent = random.choice(self.agents)
        return agent.handle_query(query)

# Usage
support_system = CustomerSupportSystem()
queries = [
    "How do I reset my password?",
    "I have a question about my recent bill",
    "What are the features of your premium plan?"
]

for query in queries:
    response = support_system.route_query(query)
    print(response)

Challenges in LLM Multi-Agent Systems

Implementing multi-agent LLM systems comes with several challenges, including coordination overhead, potential conflicts between agents, and the need for efficient resource allocation. Addressing these challenges is crucial for building effective and scalable systems.

import time
import random

class Agent:
    def __init__(self, name):
        self.name = name

    def process_task(self, task):
        processing_time = random.uniform(0.1, 0.5)
        time.sleep(processing_time)
        return f"{self.name} completed '{task}' in {processing_time:.2f} seconds"

def measure_overhead(num_agents, num_tasks):
    agents = [Agent(f"Agent{i}") for i in range(num_agents)]
    tasks = [f"Task{i}" for i in range(num_tasks)]

    start_time = time.time()
    for task in tasks:
        agent = random.choice(agents)
        result = agent.process_task(task)
        print(result)

    total_time = time.time() - start_time
    print(f"Total time: {total_time:.2f} seconds")
    print(f"Average time per task: {total_time/num_tasks:.2f} seconds")

# Measure overhead for different system sizes
measure_overhead(num_agents=3, num_tasks=10)
measure_overhead(num_agents=10, num_tasks=20)

Scalability and Load Balancing

As the number of agents and tasks increases, efficient load balancing becomes crucial. Implementing dynamic task allocation and agent scaling helps maintain system performance under varying workloads.

import random
import time

class LoadBalancer:
    def __init__(self, agents):
        self.agents = agents
        self.task_counts = {agent: 0 for agent in agents}

    def get_least_busy_agent(self):
        return min(self.task_counts, key=self.task_counts.get)

    def assign_task(self, task):
        agent = self.get_least_busy_agent()
        self.task_counts[agent] += 1
        return agent.process_task(task)

def simulate_workload(load_balancer, num_tasks):
    start_time = time.time()
    for i in range(num_tasks):
        task = f"Task{i}"
        result = load_balancer.assign_task(task)
        print(result)
    
    total_time = time.time() - start_time
    print(f"Processed {num_tasks} tasks in {total_time:.2f} seconds")
    print(f"Task distribution: {load_balancer.task_counts}")

# Create agents and load balancer
agents = [Agent(f"Agent{i}") for i in range(5)]
load_balancer = LoadBalancer(agents)

# Simulate workload
simulate_workload(load_balancer, num_tasks=50)

Security and Privacy Considerations

Multi-agent LLM systems often handle sensitive information, making security and privacy crucial concerns. Implementing proper access controls, data encryption, and anonymization techniques helps protect user data and maintain system integrity.

import hashlib
from cryptography.fernet import Fernet

class SecureAgent:
    def __init__(self, name, encryption_key):
        self.name = name
        self.fernet = Fernet(encryption_key)

    def encrypt_data(self, data):
        return self.fernet.encrypt(data.encode()).decode()

    def decrypt_data(self, encrypted_data):
        return self.fernet.decrypt(encrypted_data.encode()).decode()

    def process_task(self, task):
        # Anonymize sensitive information
        anonymized_task = self.anonymize_data(task)
        
        # Process the anonymized task
        result = f"Processed: {anonymized_task}"
        
        # Encrypt the result before sending
        encrypted_result = self.encrypt_data(result)
        return encrypted_result

    def anonymize_data(self, data):
        # Simple anonymization by hashing sensitive parts
        return hashlib.sha256(data.encode()).hexdigest()[:10]

# Usage
encryption_key = Fernet.generate_key()
agent = SecureAgent("SecureAgent1", encryption_key)

sensitive_task = "Process user data: John Doe, [email protected]"
encrypted_result = agent.process_task(sensitive_task)
print(f"Encrypted result: {encrypted_result}")

decrypted_result = agent.decrypt_data(encrypted_result)
print(f"Decrypted result: {decrypted_result}")

Future Directions in LLM Multi-Agent Systems

The field of LLM multi-agent systems is rapidly evolving. Future research directions include improving agent specialization, enhancing inter-agent learning, and developing more sophisticated coordination mechanisms. These advancements will lead to more capable and efficient AI systems.

import random

class EvolvingAgent:
    def __init__(self, name, skills):
        self.name = name
        self.skills = skills
        self.experience = 0

    def perform_task(self, task):
        relevant_skill = random.choice(list(self.skills.keys()))
        success_probability = self.skills[relevant_skill] / 100
        success = random.random() < success_probability
        
        if success:
            self.experience += 1
            if self.experience % 5 == 0:
                self.improve_skills()
        
        return success

    def improve_skills(self):
        for skill in self.skills:
            self.skills[skill] = min(100, self.skills[skill] + 5)

def simulate_evolution(agents, tasks, iterations):
    for _ in range(iterations):
        agent = random.choice(agents)
        task = random.choice(tasks)
        success = agent.perform_task(task)
        print(f"{agent.name} performed {task}: {'Success' if success else 'Failure'}")

# Initialize agents and tasks
agents = [
    EvolvingAgent("Agent1", {"coding": 50, "writing": 60, "analysis": 40}),
    EvolvingAgent("Agent2", {"coding": 70, "writing": 40, "analysis": 60})
]
tasks = ["Implement feature", "Write documentation", "Analyze data"]

# Run simulation
simulate_evolution(agents, tasks, 20)

for agent in agents:
    print(f"{agent.name} final skills: {agent.skills}")

Ethical Considerations in Multi-Agent LLM Systems

As multi-agent LLM systems become more advanced, it's crucial to address ethical concerns such as bias mitigation, transparency, and accountability. Implementing fairness constraints and regularly auditing system outputs can help ensure responsible AI development.

import random

class EthicalAgent:
    def __init__(self, name, bias_threshold):
        self.name = name
        self.bias_threshold = bias_threshold
        self.decision_history = []

    def make_decision(self, input_data):
        # Simulate decision-making process
        decision = random.random()
        
        # Check for potential bias
        if self.detect_bias(decision):
            decision = self.mitigate_bias(decision)
        
        self.decision_history.append(decision)
        return decision

    def detect_bias(self, decision):
        return abs(decision - 0.5) > self.bias_threshold

    def mitigate_bias(self, decision):
        return (decision + 0.5) / 2  # Move decision closer to 0.5

    def audit_decisions(self):
        avg_decision = sum(self.decision_history) / len(self.decision_history)
        bias_level = abs(avg_decision - 0.5)
        return f"Bias level: {bias_level:.4f}"

# Create ethical agents
agents = [EthicalAgent(f"Agent{i}", bias_threshold=0.3) for i in range(3)]

# Simulate decision-making
for _ in range(100):
    for agent in agents:
        agent.make_decision(random.random())

# Audit results
for agent in agents:
    print(f"{agent.name} audit result: {agent.audit_decisions()}")

Additional Resources

For those interested in delving deeper into LLM Multi-Agent Architecture, here are some valuable resources:

1. "Multi-Agent Reinforcement Learning: An Overview" by L. Busoniu et al. (2010) ArXiv: https://arxiv.org/abs/1011.3071
2. "A Survey of Multi-Agent Systems: Coordination, Organization, and Learning" by Y. Shoham and K. Leyton-Brown (2009) ArXiv: https://arxiv.org/abs/0902.3938
3. "Challenges and Opportunities in Multi-Agent Learning for the Real World" by J.Z. Leibo et al. (2021) ArXiv: https://arxiv.org/abs/2103.11471

These papers provide comprehensive overviews and insights into the field of multi-agent systems and their applications in AI and machine learning.