multi-agent-group

multi-agent-group

#!/usr/bin/env python3
"""
Multi-agent group chat powered by evot headless.

This is a standalone Python coordinator. It does NOT run an evot server and does
NOT modify evot itself. Each agent is an independent `evot -p` headless process
with its own persistent session id. All agents see the full group context.

Core model:
  - One human participant.
  - N AI agents.
  - A simple external loop asks agents to respond.
  - Agents can see and react to each other.
  - Agents can choose PASS when they have nothing to say.

Usage:
  python3 group.py
  python3 group.py -c roles.toml
"""

from __future__ import annotations

import argparse
import asyncio
import os
import re
import shutil
import sys
import textwrap
import uuid
from dataclasses import dataclass, field
from datetime import datetime
from pathlib import Path
from typing import Literal

try:
    import tomllib
except ImportError:  # Python < 3.11
    tomllib = None


# ---------------------------------------------------------------------------
# Terminal colors
# ---------------------------------------------------------------------------

class C:
    RESET = "\033[0m"
    BOLD = "\033[1m"
    DIM = "\033[2m"
    GREEN = "\033[32m"
    CYAN = "\033[36m"
    MAGENTA = "\033[35m"
    YELLOW = "\033[33m"
    BLUE = "\033[34m"
    RED = "\033[31m"
    WHITE = "\033[37m"
    BG_HUMAN = "\033[48;5;224m"


ROLE_COLORS = [C.CYAN, C.MAGENTA, C.YELLOW, C.BLUE, C.GREEN, C.WHITE]
EVOT_BIN = os.environ.get("EVOT_BIN", "evot")


# ---------------------------------------------------------------------------
# Data model
# ---------------------------------------------------------------------------

@dataclass
class AgentSpec:
    name: str
    title: str
    persona: str
    avatar: str = ""
    session_id: str = field(default_factory=lambda: f"agent-{uuid.uuid4().hex[:10]}")
    color: str = ""

    def system_prompt(self, roster: str) -> str:
        return textwrap.dedent(f"""
        你是 {self.name}，身份是 {self.title}。

        你的职责 / 人设：
        {self.persona}

        你正在一个多人群聊中。群聊里只有一个人类 Human，其他成员都是 AI agent。
        你能看到完整群聊记录；其他 agent 也能看到你的消息。

        群聊成员：
        {roster}

        群聊协议：
        - Human 是唯一指挥者。严格遵守 Human 的指令。
        - 你只能代表你自己发言，不能替其他 agent 发言、不能预告其他 agent 应该说什么。
        - 如果轮到你发言，只输出你的消息内容，不要加名字前缀。
        - 如果当前上下文不需要你发言，回复且只回复：PASS
        - 如果 Human 限制了输出格式（例如：只说一个字、只说一句话、不要解释），必须严格遵守。
        - 如果任务要求按顺序、轮流、挨个完成，你需要根据群聊记录判断当前是否轮到你；轮到你就只完成你自己的那一小部分，不要多做。
        - 可以用 @Name 提及其他 agent，但不要指挥他们，除非 Human 明确要求你协调。
        """).strip()


@dataclass
class ChatMessage:
    sender: str
    kind: Literal["human", "agent", "system"]
    content: str
    title: str = ""
    reply_to: int | None = None
    timestamp: str = field(default_factory=lambda: datetime.now().strftime("%H:%M:%S"))


@dataclass
class GroupConfig:
    max_rounds: int = 6
    context_limit: int = 80
    auto_continue: bool = True
    show_pass: bool = True


# ---------------------------------------------------------------------------
# evot headless runner
# ---------------------------------------------------------------------------

def clean_env() -> dict[str, str]:
    env = os.environ.copy()
    # Local/headless CLI calls should not fail due to SOCKS proxy envs.
    for key in ["http_proxy", "https_proxy", "all_proxy", "HTTP_PROXY", "HTTPS_PROXY", "ALL_PROXY"]:
        env.pop(key, None)
    return env


async def call_evot(prompt: str, system_prompt: str, session_id: str) -> str:
    cmd = [
        EVOT_BIN,
        "-p", prompt,
        "--output-format", "text",
        "--append-system-prompt", system_prompt,
        "--resume", session_id,
    ]

    proc = await asyncio.create_subprocess_exec(
        *cmd,
        stdout=asyncio.subprocess.PIPE,
        stderr=asyncio.subprocess.PIPE,
        env=clean_env(),
    )
    stdout, stderr = await proc.communicate()

    if proc.returncode != 0:
        err = stderr.decode(errors="replace").strip()
        return f"[evot error] {err[:300]}"

    return stdout.decode(errors="replace").strip()


# ---------------------------------------------------------------------------
# Group chat engine
# ---------------------------------------------------------------------------

class GroupChat:
    def __init__(self, agents: list[AgentSpec], config: GroupConfig | None = None):
        self.agents = agents
        self.config = config or GroupConfig()
        self.history: list[ChatMessage] = []
        self.agent_by_name = {a.name.lower(): a for a in agents}

        for i, agent in enumerate(self.agents):
            agent.color = ROLE_COLORS[i % len(ROLE_COLORS)]

    # -- context -------------------------------------------------------------

    def roster(self) -> str:
        lines = ["- Human: the only human commander"]
        for agent in self.agents:
            lines.append(f"- @{agent.name}: {agent.title} — {agent.persona}")
        return "\n".join(lines)

    def format_context(self) -> str:
        recent = self.history[-self.config.context_limit:]
        if not recent:
            return "(no messages yet)"

        lines: list[str] = []
        for i, msg in enumerate(recent, 1):
            label = "Human" if msg.kind == "human" else msg.sender
            title = f" ({msg.title})" if msg.title else ""
            if msg.reply_to is not None:
                lines.append(f"[{i}] {label}{title} reply_to={msg.reply_to}: {msg.content}")
            else:
                lines.append(f"[{i}] {label}{title}: {msg.content}")
        return "\n".join(lines)

    def mentions_in(self, text: str) -> list[AgentSpec]:
        names = re.findall(r"@(\w+)", text)
        result: list[AgentSpec] = []
        seen: set[str] = set()
        for name in names:
            key = name.lower()
            if key in self.agent_by_name and key not in seen:
                result.append(self.agent_by_name[key])
                seen.add(key)
        return result

    # -- user input ----------------------------------------------------------

    def add_human_message(self, content: str):
        self.history.append(ChatMessage(sender="Human", kind="human", content=content, title="HUMAN"))

    # -- agent prompting -----------------------------------------------------

    def build_agent_prompt(self, agent: AgentSpec) -> str:
        latest = self.history[-1] if self.history else None
        latest_text = latest.content if latest else ""
        mentioned = agent in self.mentions_in(latest_text)
        index = self.agents.index(agent) + 1
        order = " → ".join(a.name for a in self.agents)

        return textwrap.dedent(f"""
        === 群聊记录 ===
        {self.format_context()}

        === 当前轮到你 ===
        你是 @{agent.name}（{agent.title}）。
        你在 agent 顺序中排第 {index}/{len(self.agents)} 位。
        agent 顺序：{order}
        最新消息是否 @ 了你：{"是" if mentioned else "否"}

        请根据完整上下文判断是否应该发言。
        - 如果不该发言，只输出：PASS
        - 如果应该发言，只输出你的群聊消息内容
        - 你只能代表 @{agent.name} 自己发言
        - 不要替其他 agent 发言，不要把多个 agent 的内容合并到你的回复里
        - 如果 Human 要求每个人只说一个字，你就只能输出一个字
        """).strip()

    async def ask_agent(self, agent: AgentSpec) -> str | None:
        prompt = self.build_agent_prompt(agent)
        reply = await call_evot(prompt, agent.system_prompt(self.roster()), agent.session_id)
        cleaned = normalize_reply(reply)
        if is_pass(cleaned):
            return None
        return cleaned

    # -- loop strategies -----------------------------------------------------

    async def run_once(self, agents: list[AgentSpec] | None = None) -> int:
        """Ask each selected agent once. Return number of agents who spoke."""
        spoke = 0
        selected = agents or self.agents

        for agent in selected:
            print_status(f"{agent.name} thinking")
            reply = await self.ask_agent(agent)
            clear_status()

            if reply is None:
                if self.config.show_pass:
                    print_pass(agent)
                continue

            self.history.append(ChatMessage(
                sender=agent.name,
                kind="agent",
                title=agent.title,
                content=reply,
            ))
            print_agent(agent, reply)
            spoke += 1

        return spoke

    async def run_until_quiet(self) -> None:
        """Round-robin until every agent passes or max_rounds is hit."""
        for round_no in range(1, self.config.max_rounds + 1):
            print_dim(f"── round {round_no} ──")
            spoke = await self.run_once()
            if spoke == 0:
                print_dim("everyone passed; conversation is quiet")
                print()
                return
        print_dim(f"max rounds reached ({self.config.max_rounds})")
        print()

    async def run_mentions_first(self) -> None:
        """If latest message @mentions agents, ask them first, then optionally continue."""
        latest = self.history[-1].content if self.history else ""
        mentioned = self.mentions_in(latest)
        if mentioned:
            spoke = await self.run_once(mentioned)
            if spoke > 0 and self.config.auto_continue:
                await self.run_until_quiet()
            return
        await self.run_until_quiet()

    # -- maintenance ---------------------------------------------------------

    def clear(self):
        self.history.clear()
        for agent in self.agents:
            agent.session_id = f"agent-{uuid.uuid4().hex[:10]}"


# ---------------------------------------------------------------------------
# Reply normalization
# ---------------------------------------------------------------------------

def normalize_reply(reply: str) -> str:
    text = reply.strip()
    # Some models wrap PASS or replies in quotes.
    if len(text) >= 2 and ((text[0] == text[-1] == '"') or (text[0] == text[-1] == "'")):
        text = text[1:-1].strip()
    return text


def is_pass(reply: str) -> bool:
    upper = reply.strip().upper()
    return upper == "PASS" or upper.startswith("PASS\n") or upper.startswith("[PASS]")


# ---------------------------------------------------------------------------
# Rendering
# ---------------------------------------------------------------------------

def print_dim(text: str):
    print(f"  {C.DIM}{text}{C.RESET}")


def print_status(text: str):
    print(f"  {C.DIM}{text}...{C.RESET}", end="\r", flush=True)


def clear_status():
    print(" " * 80, end="\r", flush=True)


def print_pass(agent: AgentSpec):
    print(f"  {C.DIM}[{agent.name}] pass{C.RESET}")


def print_agent(agent: AgentSpec, content: str):
    role = f" {agent.title}" if agent.title else ""
    print(f"{agent.color}{C.BOLD}{agent.name}{C.RESET}{C.DIM}{role}{C.RESET}")

    lines = content.splitlines() or [""]
    if len(lines) == 1 and len(lines[0]) <= 80:
        print(f"  {agent.color}>{C.RESET} {lines[0]}")
    else:
        for line in lines:
            print(f"  {agent.color}│{C.RESET} {line}")
    print()


def print_human(content: str):
    print(f"{C.GREEN}{C.BOLD}Human >{C.RESET} {content}")


def print_banner(chat: GroupChat):
    print(f"""
{C.BOLD}╔══════════════════════════════════════════════════════════╗
║                  Multi-Agent Group Chat                 ║
╚══════════════════════════════════════════════════════════╝{C.RESET}
""")
    print(f"{C.BOLD}Agents:{C.RESET}")
    for agent in chat.agents:
        print(f"  {agent.color}● {agent.name}{C.RESET} {C.DIM}{agent.title} — {agent.persona}{C.RESET}")
    print()
    print(f"Type a message. Commands: {C.BOLD}/help{C.RESET}, {C.BOLD}/quit{C.RESET}")
    print()


HELP = f"""
{C.BOLD}Commands{C.RESET}
  <message>          Human message; agents react until quiet
  /once <message>    Human message; only one round of agents
  /ask Name <msg>    Ask one agent directly
  /roles             Show agents
  /history           Show transcript
  /clear             Clear transcript and reset agent sessions
  /rounds <n>        Set max auto rounds
  /pass on|off       Show/hide PASS lines
  /help              Show help
  /quit              Exit

{C.BOLD}Tips{C.RESET}
  - Use @Name to route to a specific agent.
  - Agents can @mention each other.
  - All agents see the full transcript.
  - There is no hidden coordinator; only Human commands the group.
"""


# ---------------------------------------------------------------------------
# Config loading
# ---------------------------------------------------------------------------

DEFAULT_AGENTS = [
    AgentSpec(
        name="Nova",
        title="PRODUCT MANAGER",
        persona="Owns product direction, requirements, tradeoffs, and task framing.",
    ),
    AgentSpec(
        name="Bram",
        title="ENGINEER",
        persona="Owns technical implementation, architecture, estimates, and risks.",
    ),
    AgentSpec(
        name="Iris",
        title="DESIGNER",
        persona="Owns UX, interaction design, clarity, and user-facing details.",
    ),
    AgentSpec(
        name="Atlas",
        title="RESEARCHER",
        persona="Owns research, evidence, edge cases, and comparative analysis.",
    ),
]


def load_agents(path: str | None) -> list[AgentSpec]:
    if path is None:
        return DEFAULT_AGENTS
    if tomllib is None:
        print("Python 3.11+ is required to load TOML config.")
        sys.exit(1)

    with open(path, "rb") as f:
        data = tomllib.load(f)

    agents = []
    for item in data.get("agents", data.get("roles", [])):
        persona = item.get("persona", item.get("description", ""))
        extra = item.get("system_prompt", "")
        if extra:
            persona = f"{persona}\n{extra}" if persona else extra
        agents.append(AgentSpec(
            name=item["name"],
            title=item.get("title", item.get("role", "AGENT")),
            persona=persona,
            avatar=item.get("avatar", ""),
        ))
    return agents


# ---------------------------------------------------------------------------
# REPL
# ---------------------------------------------------------------------------

async def repl(chat: GroupChat):
    print_banner(chat)

    while True:
        try:
            raw = input(f"{C.GREEN}{C.BOLD}You >{C.RESET} ").strip()
        except (EOFError, KeyboardInterrupt):
            print("\nbye")
            break

        if not raw:
            continue

        if raw in {"/quit", "/exit"}:
            break
        if raw == "/help":
            print(HELP)
            continue
        if raw == "/roles":
            print_banner(chat)
            continue
        if raw == "/history":
            print(chat.format_context())
            print()
            continue
        if raw == "/clear":
            chat.clear()
            print_dim("history cleared; sessions reset")
            continue
        if raw.startswith("/rounds "):
            try:
                chat.config.max_rounds = int(raw.split(maxsplit=1)[1])
                print_dim(f"max rounds = {chat.config.max_rounds}")
            except ValueError:
                print(f"{C.RED}invalid number{C.RESET}")
            continue
        if raw.startswith("/pass "):
            val = raw.split(maxsplit=1)[1].strip().lower()
            chat.config.show_pass = val in {"on", "true", "1", "yes"}
            print_dim(f"show pass = {chat.config.show_pass}")
            continue
        if raw.startswith("/ask "):
            m = re.match(r"/ask\s+(\w+)\s+(.+)", raw, re.DOTALL)
            if not m:
                print(f"{C.RED}usage: /ask Name message{C.RESET}")
                continue
            name, msg = m.group(1), m.group(2).strip()
            agent = chat.agent_by_name.get(name.lower())
            if not agent:
                print(f"{C.RED}unknown agent: {name}{C.RESET}")
                continue
            chat.add_human_message(f"@{agent.name} {msg}")
            await chat.run_once([agent])
            continue
        if raw.startswith("/once "):
            msg = raw.split(maxsplit=1)[1].strip()
            chat.add_human_message(msg)
            await chat.run_once()
            continue

        chat.add_human_message(raw)
        await chat.run_mentions_first()


# ---------------------------------------------------------------------------
# Main
# ---------------------------------------------------------------------------

def main():
    parser = argparse.ArgumentParser(description="Multi-agent group chat using evot headless")
    parser.add_argument("-c", "--config", help="TOML config file")
    parser.add_argument("--rounds", type=int, default=6, help="max auto rounds")
    parser.add_argument("--hide-pass", action="store_true", help="hide PASS lines")
    args = parser.parse_args()

    if not shutil.which(EVOT_BIN):
        print(f"{C.RED}error: '{EVOT_BIN}' not found. Install evot: https://github.com/evotai/evot{C.RESET}")
        sys.exit(1)

    agents = load_agents(args.config)
    if not agents:
        print(f"{C.RED}error: no agents configured{C.RESET}")
        sys.exit(1)

    chat = GroupChat(
        agents=agents,
        config=GroupConfig(max_rounds=args.rounds, show_pass=not args.hide_pass),
    )
    asyncio.run(repl(chat))


if __name__ == "__main__":
    main()

roles.toml

# 自定义角色配置示例

[[roles]]
name = "Developer"
description = "负责根据任务卡片完成代码开发、问题修复和技术自测。"
system_prompt = "你擅长 Rust/TypeScript，代码风格简洁，注重性能。"

[[roles]]
name = "QA"
description = "负责功能测试、问题记录和最终验收，确保功能符合 PRD 和验收标准。"
system_prompt = "你对边界条件非常敏感，善于发现潜在 bug。"

[[roles]]
name = "Cindy"
description = "Onboarding Assistant，负责新人引导、文档维护和流程答疑。"
system_prompt = "你善于用简单的语言解释复杂概念，注重文档可读性。"

[[roles]]
name = "Leader"
description = "负责团队管理、需求澄清、方案设计和任务拆分，推动功能从需求到验收完整落地。"
system_prompt = "你关注全局，擅长拆解复杂问题，优先级清晰。"

python3 group.py

╬══════════════════════════════════════════════════════════╗
║          多角色群聊协调器  (coordinator loop)        ║
╚══════════════════════════════════════════════════════════╝


  角色列表:
  ● Developer  负责根据任务卡片完成代码开发、问题修复和技术自测。
  ● QA  负责功能测试、问题记录和最终验收，确保功能符合 PRD 和验收标准。
  ● Cindy  Onboarding Assistant，负责新人引导、文档维护和流程答疑。
  ● Leader  负责团队管理、需求澄清、方案设计和任务拆分，推动功能从需求到验收完整落地。
  ○ Coordinator  (隐藏) 负责调度，决定找谁做什么

  输入任务描述即可开始，输入 /help 查看命令

You > 大家好啊，给你们一个任务，挨个说出：别逗你爻光姐笑了 这句话的每一个字，要求必须按照顺序一个个说出来

  任务开始，coordinator 接管调度...

  [1] @Leader: 请按顺序说出"别逗你爻光姐笑了"这句话的前两个字：别、逗（每个字单独一行）
  Leader >
  │ 别
  │ 逗

  [2] @Developer: 请按顺序说出"别逗你爻光姐笑了"这句话的第三、四个字：你、爻（每个字单独一行）
  Developer >
  │ 你
  │ 爻

  [3] @QA: 请按顺序说出"别逗你爻光姐笑了"这句话的第五、六个字：光、姐（每个字单独一行）
  QA >
  │ 光
  │ 姐

  [4] @Cindy: 请按顺序说出"别逗你爻光姐笑了"这句话的第七、八个字：笑、了（每个字单独一行）
  Cindy >
  │ 笑
  │ 了

  [5] ✓ 任务完成。四位角色已按顺序逐字说出"别逗你爻光姐笑了"：Leader（别、逗）→ Developer（你、爻）→ QA（光、姐）→ Cindy（笑、了）。

You > 大家好啊，给你们一个任务，每个人挨个说出：别逗你爻光姐笑了 这句话的每一个字，要求必须按照顺序一个个说出来，每个人只允许说一个字。

  任务开始，coordinator 接管调度...

  [1] @Leader: 请说出"别逗你爻光姐笑了"这句话的第一个字：别
  Leader > 别

  [2] @Developer: 请说出"别逗你爻光姐笑了"这句话的第二个字：逗
  Developer > 逗

  [3] @QA: 请说出"别逗你爻光姐笑了"这句话的第三个字：你
  QA > 你

  [4] @Cindy: 请说出"别逗你爻光姐笑了"这句话的第四个字：爻
  Cindy > 爻

  [5] @Leader: 请说出"别逗你爻光姐笑了"这句话的第五个字：光
  Leader > 光

  [6] @Developer: 请说出"别逗你爻光姐笑了"这句话的第六个字：姐
  Developer > 姐

  [7] @QA: 请说出"别逗你爻光姐笑了"这句话的第七个字：笑
  QA > 笑

  [8] @Cindy: 请说出"别逗你爻光姐笑了"这句话的第八个字：了
  Cindy > 了

  [9] ✓ 任务完成。四位角色轮流每人每次只说一个字，按顺序完成了"别逗你爻光姐笑了"：Leader（别）→ Developer（逗）→ QA（你）→ Cindy（爻）→ Leader（光）→ Developer（姐）→ QA（笑）→ Cindy（了）。