141 lines
4.4 KiB
Python
141 lines
4.4 KiB
Python
"""
|
|
Agent 基类 - 定义 Agent 的基本行为
|
|
"""
|
|
from dataclasses import dataclass
|
|
from typing import Generator
|
|
from agents.agent_profiles import get_agent_profile
|
|
|
|
|
|
@dataclass
|
|
class AgentMessage:
|
|
"""Agent 发言消息"""
|
|
agent_id: str
|
|
agent_name: str
|
|
emoji: str
|
|
content: str
|
|
round_num: int
|
|
|
|
|
|
class BaseAgent:
|
|
"""Agent 基类"""
|
|
|
|
def __init__(self, agent_id: str, llm_client, language: str = "Chinese"):
|
|
"""
|
|
初始化 Agent
|
|
|
|
Args:
|
|
agent_id: Agent 标识符 (如 'ceo', 'cto')
|
|
llm_client: LLM 客户端实例
|
|
language: 输出语言
|
|
"""
|
|
self.agent_id = agent_id
|
|
self.llm_client = llm_client
|
|
self.language = language
|
|
|
|
profile = get_agent_profile(agent_id)
|
|
if not profile:
|
|
raise ValueError(f"未知的 Agent ID: {agent_id}")
|
|
|
|
self.name = profile["name"]
|
|
self.emoji = profile["emoji"]
|
|
self.perspective = profile["perspective"]
|
|
self.focus_areas = profile["focus_areas"]
|
|
self.system_prompt = f"{profile['system_prompt']}\n\nIMPORTANT: You MUST output your response in {self.language}."
|
|
|
|
# 存储对话历史
|
|
self.conversation_history = []
|
|
|
|
@property
|
|
def model_name(self) -> str:
|
|
"""获取当前使用的模型名称"""
|
|
if hasattr(self.llm_client, "model"):
|
|
return self.llm_client.model
|
|
return "Unknown Model"
|
|
|
|
def generate_response(
|
|
self,
|
|
topic: str,
|
|
context: str = "",
|
|
previous_speeches: list = None,
|
|
round_num: int = 1
|
|
) -> Generator[str, None, None]:
|
|
"""
|
|
生成 Agent 的发言（流式输出）
|
|
|
|
Args:
|
|
topic: 讨论议题
|
|
context: 背景信息
|
|
previous_speeches: 之前其他 Agent 的发言列表
|
|
round_num: 当前轮次
|
|
|
|
Yields:
|
|
str: 流式输出的文本片段
|
|
"""
|
|
# 构建对话 prompt
|
|
user_prompt = self._build_user_prompt(topic, context, previous_speeches, round_num)
|
|
|
|
# 调用 LLM 生成回复
|
|
full_response = ""
|
|
for chunk in self.llm_client.chat_stream(
|
|
system_prompt=self.system_prompt,
|
|
user_prompt=user_prompt
|
|
):
|
|
full_response += chunk
|
|
yield chunk
|
|
|
|
# 保存到历史
|
|
self.conversation_history.append({
|
|
"round": round_num,
|
|
"content": full_response
|
|
})
|
|
|
|
def _build_user_prompt(
|
|
self,
|
|
topic: str,
|
|
context: str,
|
|
previous_speeches: list,
|
|
round_num: int
|
|
) -> str:
|
|
"""构建用户 prompt"""
|
|
|
|
prompt_parts = [f"## 讨论议题\n{topic}"]
|
|
|
|
if context:
|
|
prompt_parts.append(f"\n## 背景信息\n{context}")
|
|
|
|
if previous_speeches and len(previous_speeches) > 0:
|
|
prompt_parts.append("\n## 其他人的观点")
|
|
for speech in previous_speeches:
|
|
prompt_parts.append(
|
|
f"\n**{speech['emoji']} {speech['name']}**:\n{speech['content']}"
|
|
)
|
|
|
|
if round_num == 1:
|
|
prompt_parts.append(
|
|
f"\n## 你的任务\n"
|
|
f"作为 {self.name}，请从你的专业视角（{self.perspective}）对这个议题发表看法。\n"
|
|
f"重点关注：{', '.join(self.focus_areas)}\n"
|
|
f"请给出 2-3 个核心观点，每个观点用 1-2 句话阐述。保持简洁有力。"
|
|
)
|
|
else:
|
|
prompt_parts.append(
|
|
f"\n## 你的任务\n"
|
|
f"这是第 {round_num} 轮讨论。请针对其他人的观点进行回应：\n"
|
|
f"- 你同意或反对哪些观点？为什么？\n"
|
|
f"- 有没有被忽略的重要问题？\n"
|
|
f"- 你的立场有没有调整？\n"
|
|
f"请保持简洁，聚焦于最重要的 1-2 个点。"
|
|
)
|
|
|
|
return "\n".join(prompt_parts)
|
|
|
|
def get_summary(self) -> str:
|
|
"""获取该 Agent 所有发言的摘要"""
|
|
if not self.conversation_history:
|
|
return "暂无发言"
|
|
|
|
return "\n---\n".join([
|
|
f"第 {h['round']} 轮: {h['content']}"
|
|
for h in self.conversation_history
|
|
])
|