Camel-AI: Multi-Agent Systems That Actually Collaborate Effectively

# Multi-agent setup with communication guards
from camel.agents import ChatAgent
from camel.messages import BaseMessage
from camel.conversations import Conversation
from camel.types import RoleType

# Define agents with clear objectives
pm_agent = ChatAgent(
    system_message="""You are a Product Manager.
    OBJECTIVE: Define product requirements.
    DECISION AUTHORITY: You make final decisions on features.
    COMMUNICATION STYLE: Be decisive, not exploratory.
    After 3 rounds of discussion, you MUST make a decision.
    """,
    role_type=RoleType.PRODUCT_MANAGER,
    decision_authority=True  # This agent can make final decisions
)

dev_agent = ChatAgent(
    system_message="""You are a Senior Developer.
    OBJECTIVE: Assess technical feasibility.
    DECISION AUTHORITY: You advise but PM decides.
    COMMUNICATION STYLE: Be specific about technical concerns.
    If you need clarification, ask ONCE, then make assumptions.
    """,
    role_type=RoleType.DEVELOPER,
    decision_authority=False
)

# Conversation with guards
conversation = Conversation(
    agents=[pm_agent, dev_agent],
    # Communication guards
    max_turns=6,  # Total limit
    decision_required_after=3,  # Must decide by turn 3
    # Loop detection
    detect_loops=True,
    loop_threshold=2,  # Alert if same point discussed 2x
    break_loop_action="force_decision",  # Force decision on loop
    # State tracking
    track_conversation_state=True,
    state_save_interval=2
)

# Run with monitoring
turn_count = 0
decision_reached = False

while turn_count < 6 and not decision_reached:
    # Get next agent response
    response = conversation.next_turn()

    # Check for decision
    if "DECISION:" in response.content:
        decision_reached = True
        print(f"Decision reached: {response.content}")
        break

    # Check for loops
    if conversation.is_looping():
        print("Loop detected, forcing decision")
        conversation.force_decision(agent=pm_agent)
        break

    turn_count += 1

# Role enforcement with validation
from camel.agents import ChatAgent
from camel.validators import RoleValidator

# Define role permissions
role_permissions = {
    "product_manager": {
        "allowed_actions": ["define_requirements", "prioritize_features", "make_decisions"],
        "forbidden_actions": ["write_code", "design_ui"],
        "can_decide": True,
        "must_consult": ["developer", "designer"]
    },
    "developer": {
        "allowed_actions": ["assess_feasibility", "estimate_effort", "write_code"],
        "forbidden_actions": ["define_requirements", "make_product_decisions"],
        "can_decide": False,
        "must_consult": ["product_manager"]
    },
    "designer": {
        "allowed_actions": ["design_ui", "define_user_experience", "create_mockups"],
        "forbidden_actions": ["write_code", "define_technical_specs"],
        "can_decide": False,
        "must_consult": ["product_manager"]
    }
}

# Create validator
validator = RoleValidator(role_permissions)

# Create agents with role enforcement
pm = ChatAgent(
    system_message="You are a Product Manager...",
    role="product_manager",
    # Enable role enforcement
    enforce_role=True,
    validator=validator
)

developer = ChatAgent(
    system_message="You are a Senior Developer...",
    role="developer",
    enforce_role=True,
    validator=validator
)

# Before each response, validate
def get_validated_response(agent, message):
    # Generate response
    response = agent.step(message)

    # Validate action
    validation = validator.validate_action(
        role=agent.role,
        action=response.action,
        content=response.content
    )

    if not validation.is_valid:
        # Re-generate with correction
        correction = f"Your response violated your role. {validation.error}. Stay within your authority."
        response = agent.step(message + f"\n\n{correction}")

    return response

# Now agents:
# - Cannot take actions outside their role
# - Are corrected if they drift
# - Maintain consistent character

# Workflow orchestration with stage gates
from camel.workflows import Workflow, Stage, StageGate
from camel.agents import ChatAgent

# Define agents
pm = ChatAgent("Product Manager")
dev = ChatAgent("Developer")
designer = ChatAgent("Designer")
qa = ChatAgent("QA Engineer")

# Define workflow stages
stages = [
    Stage(
        name="requirements",
        description="Define product requirements",
        required_agents=[pm],
        optional_agents=[dev, designer],
        # Stage gate
        gate=StageGate(
            requirement="Clear requirements document",
            validator=lambda msg: "REQUIREMENTS:" in msg.content,
            approver=pm
        )
    ),
    Stage(
        name="design",
        description="Design UI/UX",
        required_agents=[designer],
        optional_agents=[pm],
        # Must complete requirements first
        depends_on=["requirements"],
        gate=StageGate(
            requirement="Design mockups approved",
            validator=lambda msg: "DESIGN_APPROVED:" in msg.content,
            approver=pm
        )
    ),
    Stage(
        name="development",
        description="Implement features",
        required_agents=[dev],
        optional_agents=[designer],
        depends_on=["design"],
        gate=StageGate(
            requirement="Code implementation complete",
            validator=lambda msg: "IMPLEMENTATION_COMPLETE:" in msg.content,
            approver=dev
        )
    ),
    Stage(
        name="testing",
        description="Test and validate",
        required_agents=[qa],
        optional_agents=[dev, pm],
        depends_on=["development"],
        gate=StageGate(
            requirement="All tests passing",
            validator=lambda msg: "TESTS_PASSING:" in msg.content,
            approver=qa
        )
    )
]

# Create workflow
workflow = Workflow(
    name="web_app_development",
    stages=stages,
    # Coordination settings
    coordination_mode="sequential",  # Stages run in order
    allow_parallel=False,  # No parallel stages
    # Handoff
    handoff_method="formal",  # Formal handoff between stages
    handoff_template="""
    STAGE COMPLETE: {from_stage}
    Deliverables: {deliverables}
    Next stage: {to_stage}
    Assigned to: {next_agent}
    """,
    # Monitoring
    track_progress=True,
    save_checkpoints=True
)

# Execute workflow
result = workflow.run(
    initial_task="Build a todo list web app",
    # Callbacks
    on_stage_complete=lambda stage: print(f"Completed: {stage.name}"),
    on_gate_failure=lambda gate: print(f"Gate failed: {gate.requirement}")
)

# Workflow now:
# 1. Runs stages sequentially
# 2. Validates stage completion before proceeding
# 3. Formal handoffs between agents
# 4. No role confusion or duplication