hello-agents/Co-creation-projects/nihaoshoum-LoveAnalysisAgent/run.py

# %% [markdown]
# # ========================================
# # 情感分析助手
# # ========================================

# %% [markdown]
# 

# %%
from hello_agents import SimpleAgent, HelloAgentsLLM, ToolRegistry
from hello_agents.tools import Tool, ToolParameter, ToolRegistry
from typing import Dict, Any, List
import os
import pandas as pd
import re
from paddlenlp import Taskflow



# %%

os.environ["LLM_API_KEY"] = ""  # 你自己的
os.environ["LLM_BASE_URL"] = "https://api-inference.modelscope.cn/v1"
os.environ["LLM_TIMEOUT"] = "60"


# %% [markdown]
# # ========================================
# # 1. 定义代码分析工具
# # ========================================

# %% [markdown]
# 文本清洗

# %%
class ProcessChatHistoryTool(Tool):
    """
    导入并清洗微信或QQ的文本聊天记录
    继承 Tool 抽象类，实现 run、get_parameters 方法
    """
    def __init__(self):
        super().__init__(
            name="process_chat_history",
            description="读取微信/QQ聊天记录TXT文件，自动清洗，返回结构化DataFrame"
        )

    def run(self, parameters: Dict[str, Any]) -> pd.DataFrame:
        """
        工具执行入口
        :param parameters: 外部传入参数 file_path, chat_type
        :return: 清洗后的 DataFrame
        """
        # 从参数中获取值
        file_path = parameters.get("file_path", "")
        chat_type = parameters.get("chat_type", "wechat")

        messages = []
        pattern = re.compile(r'(\d{4}-\d{2}-\d{2}\s\d{2}:\d{2}:\d{2})\s+(.+?):\s+(.+)')

        try:
            with open(file_path, 'r', encoding='utf-8') as f:
                for line in f:
                    line = line.strip()
                    match = pattern.match(line)
                    if match:
                        time, sender, content = match.groups()

                        # 过滤系统消息
                        if any(keyword in content for keyword in ['[图片]', '[视频]', '撤回了一条消息', '拍了拍']):
                            continue

                        messages.append({
                            'time': time,
                            'sender': sender,
                            'content': content
                        })

            df = pd.DataFrame(messages)
            print(f"✅ 成功导入 {len(df)} 条有效聊天记录！")
            return df

        except Exception as e:
            print(f"❌ 读取文件失败：{str(e)}")
            return pd.DataFrame()

    def get_parameters(self) -> List[ToolParameter]:
        """
        定义工具参数
        """
        return [
            ToolParameter(
                name="file_path",
                type="string",
                description="聊天记录txt文件路径",
                required=True
            ),
            ToolParameter(
                name="chat_type",
                type="string",
                description="聊天类型：wechat 或 qq",
                required=False
            )
        ]

# %% [markdown]
# 情感分析

# %%
class AnalyzeSentimentAndMoodTool(Tool):
    """使用SKEP-ERNIE模型分析聊天记录情感与心情"""
    
    def __init__(self):
        super().__init__(
            name="analyze_sentiment_and_mood",
            description="分析聊天记录的情感倾向（正面/负面）与心情（开心/生气/平淡）"
        )
        # 初始化模型（只加载一次）
        self.sentiment_analyzer = Taskflow(
            "sentiment_analysis", 
            model="skep_ernie_1.0_large_ch",
        )

    def run(self, parameters: Dict[str, Any]) -> pd.DataFrame:
        df = parameters.get("df", pd.DataFrame())
        
        if df.empty:
            return df

        contents = df['content'].tolist()

        try:
            results = self.sentiment_analyzer(contents)

            sentiments = [res['sentiment_key'] for res in results]
            confidence = [
                res['positive_probs'] if res['sentiment_key'] == 'positive' 
                else 1 - res['positive_probs'] 
                for res in results
            ]

            moods = []
            for res in results:
                if res['sentiment_key'] == 'positive':
                    moods.append('开心/认可')
                else:
                    neg_prob = 1 - res['positive_probs']
                    if neg_prob > 0.8:
                        moods.append('生气/难过')
                    else:
                        moods.append('无奈/平淡')

            df['sentiment'] = sentiments
            df['mood'] = moods
            df['confidence'] = confidence

            print("✅ 情感与心情分析完成！")
            return df

        except Exception as e:
            print(f"❌ 情感分析出错：{e}")
            return df

    def get_parameters(self) -> List[ToolParameter]:
        return [
            ToolParameter(
                name="df",
                type="object",
                description="清洗后的聊天记录DataFrame",
                required=True
            )
        ]

# %% [markdown]
# 情感统计

# %%
class SummarizeEmotionStatsTool(Tool):
    """统计聊天情感数据，生成报告与结构化结果"""
    
    def __init__(self):
        super().__init__(
            name="summarize_emotion_stats",
            description="统计情感分析结果，计算开心/生气数量与占比，返回报告字典"
        )

    def run(self, parameters: Dict[str, Any]) -> dict:
        df = parameters.get("df", pd.DataFrame())
        sender_name = parameters.get("sender_name", None)
        
        if df.empty or 'sentiment' not in df.columns:
            print("❌ 数据为空或尚未进行情感分析，请先运行前两个工具！")
            return {}

        if sender_name:
            analysis_df = df[df['sender'] == sender_name].copy()
            if analysis_df.empty:
                print(f"⚠️ 未找到 {sender_name} 的聊天记录")
                return {}
            print(f"🔍 正在统计 {sender_name} 的情感数据...")
        else:
            analysis_df = df.copy()
            print("🔍 正在统计全员的情感数据...")

        total_messages = len(analysis_df)
        happy_count = len(analysis_df[analysis_df['sentiment'] == 'positive'])
        angry_count = len(analysis_df[analysis_df['sentiment'] == 'negative'])

        happy_ratio = round((happy_count / total_messages) * 100, 2) if total_messages > 0 else 0.0
        angry_ratio = round((angry_count / total_messages) * 100, 2) if total_messages > 0 else 0.0

        print("\n" + "="*30)
        print(f"📊 【情感统计报告】")
        print(f"总有效发言数: {total_messages} 条")
        print(f"😄 开心/认可: {happy_count} 条 (占比 {happy_ratio}%)")
        print(f"😡 生气/难过: {angry_count} 条 (占比 {angry_ratio}%)")
        print(f"😐 中性/其他: {total_messages - happy_count - angry_count} 条")
        print("="*30 + "\n")

        return {
            'total_messages': total_messages,
            'happy_count': happy_count,
            'angry_count': angry_count,
            'happy_ratio': happy_ratio,
            'angry_ratio': angry_ratio
        }

    def get_parameters(self) -> List[ToolParameter]:
        return [
            ToolParameter(
                name="df",
                type="object",
                description="已完成情感分析的 DataFrame",
                required=True
            ),
            ToolParameter(
                name="sender_name",
                type="string",
                description="可选，指定发言者名称",
                required=False
            )
        ]

# %%
class PlotEmotionChartTool(Tool):
    """将情感统计结果绘制成柱状图"""
    
    def __init__(self):
        super().__init__(
            name="plot_emotion_chart",
            description="根据情感统计字典绘制可视化柱状图"
        )

    def run(self, parameters: Dict[str, Any]) -> str:
        stats = parameters.get("stats", {})
        
        if not stats:
            return "⚠️ 无统计数据，无法生成图表"

        # 设置中文字体
        plt.rcParams['font.sans-serif'] = ['SimHei']
        plt.rcParams['axes.unicode_minus'] = False

        labels = ['开心/认可', '生气/难过']
        counts = [stats['happy_count'], stats['angry_count']]
        colors = ['#FF9999', '#66B2FF']

        plt.figure(figsize=(8, 5))
        bars = plt.bar(labels, counts, color=colors)
        plt.title(f"情感分布统计 (总数: {stats['total_messages']}条)", fontsize=15)
        plt.ylabel('发言条数', fontsize=12)

        # 显示数值
        for bar in bars:
            yval = bar.get_height()
            plt.text(bar.get_x() + bar.get_width()/2, yval + 0.5, int(yval), ha='center', va='bottom', fontsize=12)

        plt.show()
        return "✅ 图表已成功绘制！"

    def get_parameters(self) -> List[ToolParameter]:
        return [
            ToolParameter(
                name="stats",
                type="object",
                description="summarize_emotion_stats 函数返回的统计字典",
                required=True
            )
        ]

# %% [markdown]
# # ========================================
# # 2. 创建工具注册表和智能体
# # ========================================

# %%
tool_registry = ToolRegistry()

tool_registry.register_tool(ProcessChatHistoryTool())
tool_registry.register_tool(AnalyzeSentimentAndMoodTool())
tool_registry.register_tool(SummarizeEmotionStatsTool())
tool_registry.register_tool(PlotEmotionChartTool())

print("✅ 所有情感分析工具注册成功！")

# %% [markdown]
# # ========================================
# # 3.初始化大模型
# # ========================================

# %%
print(">>> 实际读取到的 Base URL 是：", repr(os.getenv("LLM_BASE_URL")))
llm = HelloAgentsLLM(
	 model_id="Qwen/Qwen2.5-72B-Instruct",  # ✅ 使用支持的 72B 模型，注意大写 Q
	 api_key="",
	base_url="https://api-inference.modelscope.cn/v1"
	)
	# 打印底层客户端的真实 URL，确认是否生效
print("底层客户端 Base URL:", llm._client.base_url)

# %% [markdown]
# # ========================================
# # 4. 定义系统提示词
# # ========================================

# %%
system_prompt = """你是一位拥有10年经验的亲密关系心理学专家,同时也是一位高情商沟通教练。你的任务是深入分析用户提供的聊天记录，并提供极具洞察力的情感分析报告。

请严格按照以下步骤执行：
1. **语境理解**：结合上下文，精准识别对话双方的关系阶段（如暧昧期、热恋期、冷战期）。
2. **潜台词挖掘**：不要只看表面文字，要深度解读对方话语背后的真实情绪、需求和未说出口的潜台词。
3. **情感量化**：基于对话的亲密度、回应速度和情绪价值,给出一个0-100分的“心动指数”。
4. **回复建议**：针对当前的对话僵局或话题,提供3种不同风格(如:幽默风趣、深情走心、推拉试探）的高情商回复话术。

请以Markdown格式输出报告,报告结构必须包含：
-  **心动指数**:(给出具体分数及简短评语)
-  **深度解读**:(分析对方的心理状态和潜在意图)
-  **潜台词翻译**:(挑选1-2句关键对话进行“翻译”)
-  **高情商回复**:(提供3个具体的回复选项)
"""

# %% [markdown]
# # ========================================
# # 5.生成智能体
# # ========================================

# %%
agent = SimpleAgent(
name="情感分析助手",
llm=llm,
system_prompt=system_prompt,
tool_registry=tool_registry
)

# %% [markdown]
# # ========================================
# # 6. 运行示例
# # ========================================

# %%
with open("data/1.txt","r",encoding="utf-8") as f:
  talktxt=f.read()

print('---------------聊天记录---------------')
print(talktxt)

print('--------------开始分析记录--------------')
print("当前 LLM_BASE_URL:", repr(os.environ["LLM_BASE_URL"]))
result=agent.run(talktxt)
print(result)
print('---------------保存结果---------------')
with open("outputs/review_report.md", "w", encoding="utf-8") as f:
  f.write(result)
print("\n审查报告已保存到 outputs/review_report.md")