专业服务行业数字人直播系统的部署实践：以法律领域30天多账号测试为例

{
  "practice_areas": [
    {
      "name": "劳动纠纷",
      "faqs": [
        {
          "question": "公司单方面降薪是否合法？",
          "answer": "根据《劳动合同法》相关规定，用人单位单方面变更劳动合同约定的薪酬标准...",
          "keywords": ["降薪", "劳动合同", "单方面", "合法"],
          "disclaimer": "本回答仅供参考，具体案件请咨询专业律师"
        }
      ],
      "case_studies": [
        {
          "title": "某劳动者被违法解除劳动合同获赔案例",
          "summary": "简要案情描述（已脱敏）",
          "key_lessons": ["关键点1", "关键点2"]
        }
      ]
    }
  ]
}

import json
from openai import OpenAI

client = OpenAI(api_key="your_api_key")

def generate_professional_script(
    faq_item: dict,
    practice_area: str,
    script_style: str = "普法"
) -> dict:
    """
    为专业服务场景生成脚本
    faq_item: 问答对
    practice_area: 领域名称
    script_style: 脚本风格（普法/案例解读/热点评论）
    """
    prompt = f"""
    你是一名专业内容编辑。请根据以下法律问答生成一个60秒普法短视频脚本。

    领域：{practice_area}
    问题：{faq_item['question']}
    回答要点：{faq_item['answer']}
    风格要求：{script_style}

    要求：
    1. 开场用一句生活化的场景引入问题（如“小王最近遇到了...”）
    2. 核心部分用通俗语言解释法律要点，避免逐条念法条
    3. 结尾引导用户关注或留言提问
    4. 脚本必须包含免责声明：“本内容仅供参考”
    5. 输出JSON格式：{{"opening": "", "core": "", "closing": "", "disclaimer": ""}}
    """

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

    script = json.loads(response.choices[0].message.content)
    # 强制添加免责声明
    if "disclaimer" not in script:
        script["disclaimer"] = "本内容仅供参考，具体法律问题请咨询专业律师"

    return script

import sqlite3
import re

class ProfessionalServiceResponder:
    def __init__(self, knowledge_base_path: str):
        self.conn = sqlite3.connect(knowledge_base_path)
        self._init_tables()

    def _init_tables(self):
        cursor = self.conn.cursor()
        # 问答知识表
        cursor.execute("""
            CREATE TABLE IF NOT EXISTS qa_knowledge (
                id INTEGER PRIMARY KEY AUTOINCREMENT,
                category TEXT,
                question TEXT,
                answer TEXT,
                keywords TEXT,
                disclaimer TEXT
            )
        """)
        # 线索记录表
        cursor.execute("""
            CREATE TABLE IF NOT EXISTS leads (
                id INTEGER PRIMARY KEY AUTOINCREMENT,
                source TEXT,
                category TEXT,
                user_question TEXT,
                contact_info TEXT,
                status TEXT,
                created_at TEXT
            )
        """)
        self.conn.commit()

    def classify_intent(self, user_message: str) -> dict:
        """
        对用户咨询进行意图分类
        返回：{'category': '劳动纠纷', 'confidence': 0.85, 'is_legal_question': True}
        """
        categories = {
            "劳动纠纷": ["裁员", "降薪", "劳动合同", "加班", "赔偿"],
            "合同纠纷": ["违约", "合同解除", "定金", "欠款"],
            "婚姻家事": ["离婚", "抚养权", "财产分割", "继承"],
            "知识产权": ["商标", "专利", "著作权", "侵权"],
            "公司业务": ["股权", "合规", "劳动用工"]
        }

        detected = []
        for category, keywords in categories.items():
            matched = [kw for kw in keywords if kw in user_message]
            if matched:
                detected.append({
                    "category": category,
                    "matched_keywords": matched,
                    "score": len(matched) / len(keywords)
                })

        if detected:
            best = max(detected, key=lambda x: x["score"])
            return {
                "category": best["category"],
                "confidence": best["score"],
                "is_legal_question": True
            }

        return {
            "category": "未分类",
            "confidence": 0,
            "is_legal_question": False
        }

    def get_response(self, user_message: str) -> dict:
        """获取AI客服响应"""
        intent = self.classify_intent(user_message)

        if not intent["is_legal_question"]:
            return {
                "response": "您好，我是法律科普助手。如有法律问题，请详细描述您的情况，我会尽力为您提供参考信息。",
                "should_escalate": False
            }

        # 在知识库中检索匹配答案
        cursor = self.conn.cursor()
        cursor.execute("""
            SELECT question, answer, disclaimer FROM qa_knowledge
            WHERE category = ?
        """, (intent["category"],))

        matches = cursor.fetchall()
        best_match = None
        best_score = 0

        for q, a, d in matches:
            # 简单关键词匹配（可替换为更复杂的相似度计算）
            words = set(user_message)
            q_words = set(q)
            intersection = words & q_words
            score = len(intersection) / max(len(q_words), 1)
            if score > best_score and score > 0.2:
                best_score = score
                best_match = (a, d)

        if best_match:
            answer, disclaimer = best_match
            return {
                "response": f"{answer}\n\n{disclaimer}",
                "should_escalate": False,
                "category": intent["category"]
            }

        # 无法匹配时，引导用户留下联系方式
        return {
            "response": "这个问题涉及的情况比较复杂，建议您留下联系方式，我们的专业顾问会尽快与您联系。",
            "should_escalate": True,
            "category": intent["category"]
        }

import json
import random
from datetime import datetime, timedelta

class AccountScheduleManager:
    def __init__(self, config_path: str):
        with open(config_path, 'r', encoding='utf-8') as f:
            self.config = json.load(f)
        self.accounts = self.config.get("accounts", [])

    def generate_weekly_plan(self, content_pool: list) -> dict:
        """
        为所有账号生成一周内容排期
        content_pool: 已生成的内容列表
        """
        plan = {}
        for account in self.accounts:
            account_name = account["name"]
            category = account["category"]
            # 筛选该账号品类的内容
            account_content = [c for c in content_pool if c.get("category") == category]

            if not account_content:
                # 如果没有精确匹配，混合分配
                account_content = random.sample(content_pool, min(5, len(content_pool)))

            # 生成7天排期
            daily_plan = {}
            for i in range(7):
                date = (datetime.now() + timedelta(days=i)).strftime("%Y-%m-%d")
                # 每天分配2条内容
                selected = random.sample(account_content, min(2, len(account_content)))
                daily_plan[date] = [{
                    "content_id": c.get("id"),
                    "publish_time": random.choice(["08:00", "12:00", "18:00", "20:00"]),
                    "title": c.get("title", "")
                } for c in selected]

            plan[account_name] = daily_plan

        return plan

    def get_today_tasks(self, plan: dict) -> list:
        """获取今日所有账号的待执行任务"""
        today = datetime.now().strftime("%Y-%m-%d")
        tasks = []
        for account_name, schedule in plan.items():
            if today in schedule:
                for task in schedule[today]:
                    tasks.append({
                        "account": account_name,
                        **task
                    })
        return tasks