个人IP数字人直播系统的部署实践：基于歌唱家账号从形象训练到单场活动的技术记录

import json
from openai import OpenAI

client = OpenAI(api_key="your_api_key")

def generate_ip_script(
    ip_persona: str,
    product_info: dict,
    engagement_hook: str = None
) -> dict:
    """
    生成个人IP风格的直播脚本
    ip_persona: IP的人设描述
    product_info: 产品信息
    engagement_hook: 互动钩子
    """
    hook_text = f"互动钩子：{engagement_hook}" if engagement_hook else ""
    
    prompt = f"""
    你是一名直播脚本策划。请为以下个人IP生成一个60秒的直播带货脚本。

    人设：{ip_persona}
    产品名称：{product_info.get('name', '')}
    产品卖点：{', '.join(product_info.get('selling_points', []))}
    产品价格：{product_info.get('price', '')}
    {hook_text}

    要求：
    1. 脚本风格需符合人设（亲切/专业/热情等）
    2. 开场用一句话建立信任或吸引注意力
    3. 核心部分结合个人经历讲述产品价值
    4. 结尾引导下单或添加联系方式
    5. 输出JSON格式：{{"opening": "", "core": "", "closing": ""}}
    """

    response = client.chat.completions.create(
        model="gpt-4o-mini",
        messages=[{"role": "user", "content": prompt}],
        response_format={"type": "json_object"},
        temperature=0.8
    )

    script = json.loads(response.choices[0].message.content)
    return script

import sqlite3
import jieba
from difflib import SequenceMatcher

class AutoResponder:
    def __init__(self, knowledge_base_path: str):
        self.conn = sqlite3.connect(knowledge_base_path)
        self._init_table()
        self._load_knowledge()

    def _init_table(self):
        cursor = self.conn.cursor()
        cursor.execute("""
            CREATE TABLE IF NOT EXISTS qa_knowledge (
                id INTEGER PRIMARY KEY AUTOINCREMENT,
                question TEXT,
                answer TEXT,
                keywords TEXT
            )
        """)
        self.conn.commit()

    def _load_knowledge(self):
        cursor = self.conn.cursor()
        cursor.execute("SELECT question, answer, keywords FROM qa_knowledge")
        self.qa_pairs = cursor.fetchall()

    def find_best_response(self, user_message: str, threshold: float = 0.3) -> str:
        """根据用户消息匹配最佳回复"""
        # 使用关键词匹配 + 相似度计算
        best_match = None
        best_score = 0

        for q, a, keywords in self.qa_pairs:
            # 关键词匹配
            keyword_list = keywords.split(',') if keywords else []
            keyword_score = sum(1 for kw in keyword_list if kw in user_message) / max(len(keyword_list), 1)

            # 文本相似度
            similarity = SequenceMatcher(None, user_message, q).ratio()

            total_score = keyword_score * 0.6 + similarity * 0.4

            if total_score > best_score and total_score > threshold:
                best_score = total_score
                best_match = a

        return best_match if best_match else "抱歉，这个问题我暂时无法回答，请留下联系方式，我们会尽快回复您。"

    def log_interaction(self, user_id: str, message: str, response: str):
        """记录互动日志"""
        cursor = self.conn.cursor()
        cursor.execute("""
            INSERT INTO interaction_logs (user_id, message, response, timestamp)
            VALUES (?, ?, ?, datetime('now'))
        """, (user_id, message, response))
        self.conn.commit()

import requests
from datetime import datetime
import sqlite3

class LiveDataCollector:
    def __init__(self, db_path: str):
        self.conn = sqlite3.connect(db_path)
        self._init_table()

    def _init_table(self):
        cursor = self.conn.cursor()
        cursor.execute("""
            CREATE TABLE IF NOT EXISTS live_metrics (
                id INTEGER PRIMARY KEY AUTOINCREMENT,
                session_id TEXT,
                start_time TEXT,
                end_time TEXT,
                duration_minutes INTEGER,
                peak_viewers INTEGER,
                avg_viewers INTEGER,
                total_interactions INTEGER,
                total_orders INTEGER,
                gmv DECIMAL(10,2)
            )
        """)
        self.conn.commit()

    def save_metrics(self, metrics: dict):
        cursor = self.conn.cursor()
        cursor.execute("""
            INSERT INTO live_metrics (
                session_id, start_time, end_time, duration_minutes,
                peak_viewers, avg_viewers, total_interactions,
                total_orders, gmv
            ) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?)
        """, (
            metrics.get('session_id'),
            metrics.get('start_time'),
            metrics.get('end_time'),
            metrics.get('duration_minutes'),
            metrics.get('peak_viewers'),
            metrics.get('avg_viewers'),
            metrics.get('total_interactions'),
            metrics.get('total_orders'),
            metrics.get('gmv')
        ))
        self.conn.commit()