251 lines
8.0 KiB
Python
251 lines
8.0 KiB
Python
#!/usr/bin/env python3
|
||
# -*- coding: utf-8 -*-
|
||
"""
|
||
八仙智能体基类
|
||
Baxian Agent Base Class
|
||
"""
|
||
|
||
import asyncio
|
||
import logging
|
||
from typing import List, Dict, Any, Optional
|
||
from datetime import datetime
|
||
|
||
from jixia_academy.core.memory_bank.interface import MemoryBankInterface
|
||
|
||
|
||
class BaxianAgent:
|
||
"""八仙智能体"""
|
||
|
||
def __init__(
|
||
self,
|
||
name: str,
|
||
personality: str,
|
||
expertise: List[str],
|
||
style: str,
|
||
memory_bank: MemoryBankInterface
|
||
):
|
||
self.name = name
|
||
self.personality = personality
|
||
self.expertise = expertise
|
||
self.style = style
|
||
self.memory_bank = memory_bank
|
||
self.initialized = False
|
||
|
||
async def initialize(self):
|
||
"""初始化智能体"""
|
||
if self.initialized:
|
||
return
|
||
|
||
print(f"🎭 初始化 {self.name} ({self.personality})")
|
||
self.initialized = True
|
||
|
||
async def close(self):
|
||
"""关闭资源"""
|
||
self.initialized = False
|
||
|
||
async def generate_viewpoint(
|
||
self,
|
||
topic: str,
|
||
history: List[Dict[str, Any]],
|
||
round_num: int
|
||
) -> str:
|
||
"""生成观点"""
|
||
|
||
# 分析历史观点
|
||
historical_analysis = await self._analyze_history(topic, history)
|
||
|
||
# 基于个性生成观点
|
||
viewpoint = await self._generate_personalized_viewpoint(
|
||
topic=topic,
|
||
historical_analysis=historical_analysis,
|
||
round_num=round_num
|
||
)
|
||
|
||
return viewpoint
|
||
|
||
async def _analyze_history(
|
||
self,
|
||
topic: str,
|
||
history: List[Dict[str, Any]]
|
||
) -> Dict[str, Any]:
|
||
"""分析历史观点"""
|
||
|
||
# 统计各角色观点
|
||
role_views = {}
|
||
for entry in history:
|
||
speaker = entry.get("speaker", "")
|
||
if speaker in ["铁拐李", "吕洞宾", "何仙姑", "张果老",
|
||
"蓝采和", "汉钟离", "韩湘子", "曹国舅"]:
|
||
if speaker not in role_views:
|
||
role_views[speaker] = []
|
||
role_views[speaker].append(entry.get("message", ""))
|
||
|
||
# 分析共识和分歧
|
||
consensus = await self._identify_consensus(role_views)
|
||
disagreements = await self._identify_disagreements(role_views)
|
||
|
||
return {
|
||
"role_views": role_views,
|
||
"consensus": consensus,
|
||
"disagreements": disagreements,
|
||
"total_entries": len(history)
|
||
}
|
||
|
||
async def _identify_consensus(self, role_views: Dict[str, List[str]]) -> List[str]:
|
||
"""识别共识"""
|
||
# 简化版共识识别
|
||
common_themes = []
|
||
|
||
# 如果多数角色提到相似的观点，视为共识
|
||
all_views = []
|
||
for views in role_views.values():
|
||
all_views.extend(views)
|
||
|
||
# 这里简化处理，实际可以使用NLP技术
|
||
if "风险" in str(all_views):
|
||
common_themes.append("关注风险控制")
|
||
if "机会" in str(all_views):
|
||
common_themes.append("看到投资机会")
|
||
|
||
return common_themes
|
||
|
||
async def _identify_disagreements(self, role_views: Dict[str, List[str]]) -> List[str]:
|
||
"""识别分歧"""
|
||
# 简化版分歧识别
|
||
disagreements = []
|
||
|
||
# 检查是否有明显相反的观点
|
||
for role, views in role_views.items():
|
||
if role != self.name:
|
||
# 这里简化处理
|
||
if "乐观" in str(views) and self.personality == "风险控制专家":
|
||
disagreements.append(f"与{role}的乐观观点存在分歧")
|
||
elif "悲观" in str(views) and self.personality == "创新思维者":
|
||
disagreements.append(f"与{role}的悲观观点存在分歧")
|
||
|
||
return disagreements
|
||
|
||
async def _generate_personalized_viewpoint(
|
||
self,
|
||
topic: str,
|
||
historical_analysis: Dict[str, Any],
|
||
round_num: int
|
||
) -> str:
|
||
"""基于个性生成观点"""
|
||
|
||
# 基于角色个性生成观点模板
|
||
templates = {
|
||
"逆向思维专家": [
|
||
"我认为当前主流观点可能存在盲区...",
|
||
"让我们从另一个角度思考这个问题...",
|
||
"我注意到一个被忽视的风险点..."
|
||
],
|
||
"理性分析者": [
|
||
"基于逻辑分析，我认为...",
|
||
"让我们用理性的眼光审视这个问题...",
|
||
"从数据角度看，情况显示..."
|
||
],
|
||
"风险控制专家": [
|
||
"我们首先需要评估潜在风险...",
|
||
"在考虑收益的同时，必须关注...",
|
||
"风险控制应该是首要考虑..."
|
||
],
|
||
"历史智慧者": [
|
||
"历史告诉我们...",
|
||
"类似的情况在历史上曾经...",
|
||
"从长期历史规律来看..."
|
||
],
|
||
"创新思维者": [
|
||
"或许我们可以尝试一种全新的方法...",
|
||
"传统的思路可能限制了我们的视野...",
|
||
"让我分享一个独特的见解..."
|
||
],
|
||
"平衡协调者": [
|
||
"让我们寻找一个平衡点...",
|
||
"各种观点都有其合理性...",
|
||
"我们需要综合考虑各方因素..."
|
||
],
|
||
"艺术感知者": [
|
||
"从美学和情感层面来看...",
|
||
"这个问题的深层含义是...",
|
||
"我感受到这个问题的艺术价值..."
|
||
],
|
||
"实务执行者": [
|
||
"让我们关注具体的实施方案...",
|
||
"实际操作中需要考虑...",
|
||
"可行的具体步骤包括..."
|
||
]
|
||
}
|
||
|
||
# 获取当前角色的模板
|
||
template = templates.get(self.personality, ["我认为..."])[0]
|
||
|
||
# 根据历史分析调整观点
|
||
adjusted_viewpoint = self._adjust_by_history(
|
||
template, historical_analysis, topic
|
||
)
|
||
|
||
return adjusted_viewpoint
|
||
|
||
def _adjust_by_history(
|
||
self,
|
||
base_template: str,
|
||
historical_analysis: Dict[str, Any],
|
||
topic: str
|
||
) -> str:
|
||
"""根据历史分析调整观点"""
|
||
|
||
# 基于历史调整观点
|
||
my_previous_views = historical_analysis["role_views"].get(self.name, [])
|
||
|
||
if my_previous_views:
|
||
# 如果之前已经发表过观点，可以深化或调整
|
||
return f"{base_template} 基于我之前的观点，我想进一步补充..."
|
||
else:
|
||
# 首次发言
|
||
return f"{base_template} 作为{self.personality}，我首先关注的是..."
|
||
|
||
def get_info(self) -> Dict[str, Any]:
|
||
"""获取智能体信息"""
|
||
return {
|
||
"name": self.name,
|
||
"personality": self.personality,
|
||
"expertise": self.expertise,
|
||
"style": self.style,
|
||
"initialized": self.initialized
|
||
}
|
||
|
||
|
||
async def test_baxian_agent():
|
||
"""测试八仙智能体"""
|
||
from jixia_academy.core.memory_bank.factory import get_memory_backend
|
||
|
||
memory_bank = get_memory_backend()
|
||
await memory_bank.initialize()
|
||
|
||
# 测试单个智能体
|
||
agent = BaxianAgent(
|
||
name="铁拐李",
|
||
personality="逆向思维专家",
|
||
expertise=["批判性思维", "风险识别", "逆向投资"],
|
||
style="直接犀利，善于质疑",
|
||
memory_bank=memory_bank
|
||
)
|
||
|
||
await agent.initialize()
|
||
|
||
# 测试观点生成
|
||
viewpoint = await agent.generate_viewpoint(
|
||
topic="人工智能对投资的影响",
|
||
history=[],
|
||
round_num=1
|
||
)
|
||
|
||
print(f"\n🎯 铁拐李的观点: {viewpoint}")
|
||
|
||
await agent.close()
|
||
await memory_bank.close()
|
||
|
||
|
||
if __name__ == "__main__":
|
||
asyncio.run(test_baxian_agent()) |