AutoGen多Agent对话：协作模式

import asynciofrom typing import List, Dict, Any, Optionalfrom dataclasses import dataclass, field  @dataclassclass Message:    """对话消息"""    role: str  # agent name    content: str    token_count: int = 0  class ConversableAgent:    """AutoGen风格的可对话Agent基类"""     def __init__(self, name: str, system_message: str, tools: List[str] = None):        self.name = name        self.system_message = system_message        self.tools = tools or []     async def generate_reply(self, messages: List[Message]) -> str:        """根据对话历史生成回复"""        context = self._build_context(messages)        reply = f"[{self.name}] 基于上下文生成的回复"        return reply     def _build_context(self, messages: List[Message]) -> str:        """构建LLM输入上下文"""        context = f"System: {self.system_message}\n"        for msg in messages:            context += f"{msg.role}: {msg.content}\n"        return context  class GroupChat:    """群组对话管理器"""     def __init__(self, agents: List[ConversableAgent], max_round: int = 10,                 speaker_selection: str = "round_robin"):        self.agents = agents        self.max_round = max_round        self.speaker_selection = speaker_selection        self.messages: List[Message] = []        self._current_speaker_idx = 0     def select_speaker(self) -> ConversableAgent:        """选择下一发言Agent"""        if self.speaker_selection == "round_robin":            agent = self.agents[self._current_speaker_idx % len(self.agents)]            self._current_speaker_idx += 1            return agent        elif self.speaker_selection == "auto":            return self._llm_select_speaker()        return self.agents[0]     def _llm_select_speaker(self) -> ConversableAgent:        """LLM裁决模式：根据对话历史智能选择发言者"""        agent_descriptions = ", ".join(            f"{a.name}: {a.system_message[:50]}" for a in self.agents        )        prompt = f"Based on the conversation, who should speak next?\nAgents: {agent_descriptions}"        selected = self.agents[self._current_speaker_idx % len(self.agents)]        self._current_speaker_idx += 1        return selected     def compress_history(self, max_window: int = 5) -> List[Message]:        """滑动窗口压缩对话历史"""        if len(self.messages) <= max_window:            return self.messages        summary_content = f"[摘要] 前{len(self.messages) - max_window}轮对话的关键决策已记录"        summary_msg = Message(role="system", content=summary_content)        recent = self.messages[-max_window:]        return [summary_msg] + recent     async def run(self, initial_message: str = "") -> List[Message]:        """运行群组对话"""        if initial_message:            self.messages.append(Message(role="user", content=initial_message))         for round_num in range(self.max_round):            speaker = self.select_speaker()            compressed = self.compress_history(max_window=5)            reply = await speaker.generate_reply(compressed)            msg = Message(role=speaker.name, content=reply)            self.messages.append(msg)             if "TERMINATE" in reply:                break         return self.messages  async def main():    """示例：Coder-Reviewer-Tester三人协作"""    coder = ConversableAgent("Coder", "你是Python开发者，输出可运行代码")    reviewer = ConversableAgent("Reviewer", "你是代码审查专家，检查安全性和规范")    tester = ConversableAgent("Tester", "你是测试工程师，编写pytest测试用例")     group_chat = GroupChat(        agents=[coder, reviewer, tester],        max_round=8,        speaker_selection="round_robin"    )    messages = await group_chat.run("请实现一个快速排序算法")    print(f"对话完成，共{len(messages)}条消息")  if __name__ == "__main__":    asyncio.run(main())