单人部署法律垂类AI数字人直播系统的技术实践：从零搭建9个垂直账号的60天记录

{
  "topics": [
    {
      "category": "劳动纠纷",
      "question": "公司单方面降薪是否合法",
      "answer": "根据《劳动合同法》相关规定，用人单位单方面变更劳动合同内容...",
      "keywords": ["降薪", "劳动合同", "单方面"],
      "risk_level": "中"
    }
  ]
}

import json
import random
from openai import OpenAI

client = OpenAI(api_key="your_api_key")

def load_knowledge_base(file_path: str) -> dict:
    with open(file_path, 'r', encoding='utf-8') as f:
        return json.load(f)

def select_topic_for_script(kb: dict, previous_topics: list) -> dict:
    """从知识库中选题，避免短期重复"""
    available = [t for t in kb["topics"] if t["question"] not in previous_topics]
    if not available:
        return random.choice(kb["topics"])
    return random.choice(available)

def generate_legal_script(topic: dict, platform: str = "general") -> dict:
    """
    生成法律科普短视频脚本
    platform参数用于适配不同平台的内容规范
    """
    prompt = f"""
    你是一名法律科普内容编辑。请根据以下法律问答，生成一个60秒短视频脚本。

    问题：{topic['question']}
    回答要点：{topic['answer']}
    风险等级：{topic['risk_level']}

    要求：
    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.75
    )

    script = json.loads(response.choices[0].message.content)
    script["topic"] = topic["question"]
    script["risk_level"] = topic["risk_level"]
    return script

import numpy as np
from sklearn.feature_extraction.text import TfidfVectorizer
from sklearn.metrics.pairwise import cosine_similarity

class LegalKnowledgeRetriever:
    def __init__(self, knowledge_base: dict):
        self.kb = knowledge_base
        self.vectorizer = TfidfVectorizer()
        self.corpus = [item["question"] + " " + item["answer"] for item in self.kb["topics"]]
        self.tfidf_matrix = self.vectorizer.fit_transform(self.corpus)

    def retrieve(self, query: str, top_k: int = 3) -> list:
        """根据用户问题检索最相关的法律知识条目"""
        query_vec = self.vectorizer.transform([query])
        similarities = cosine_similarity(query_vec, self.tfidf_matrix).flatten()
        top_indices = np.argsort(similarities)[-top_k:][::-1]

        results = []
        for idx in top_indices:
            if similarities[idx] > 0.1:
                results.append(self.kb["topics"][idx])
        return results

    def generate_script_with_context(self, user_query: str) -> dict:
        """检索增强的脚本生成"""
        relevant = self.retrieve(user_query)
        if not relevant:
            return {"error": "未找到相关法律知识，请扩展知识库"}

        context = "\n".join([f"问题：{r['question']}\n回答：{r['answer']}" for r in relevant])
        prompt = f"""
        基于以下法律知识内容，生成一个60秒短视频脚本：
        {context}

        要求同上（开场、核心、结论三段式，包含免责说明）
        """

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

        return {"script": response.choices[0].message.content, "sources": relevant}

import configparser
from datetime import datetime

class AccountManager:
    def __init__(self, config_path: str):
        self.config = configparser.ConfigParser()
        self.config.read(config_path, encoding='utf-8')
        self.accounts = self.config.sections()
        self.current_index = 0
        self.last_post_time = {}

    def get_next_account(self) -> dict:
        """轮询获取下一个待发内容的目标账号"""
        account_name = self.accounts[self.current_index]
        account_data = dict(self.config[account_name])
        self.current_index = (self.current_index + 1) % len(self.accounts)
        return {"name": account_name, **account_data}

    def get_distribution_plan(self, daily_capacity: int = 5) -> list:
        """生成当日分发计划，确保每个账号分配均衡"""
        plan = []
        for i in range(daily_capacity):
            account = self.get_next_account()
            today = datetime.now().strftime("%Y-%m-%d")
            key = f"{account['name']}_{today}"
            if key in self.last_post_time and self.last_post_time[key] >= 2:
                continue
            plan.append(account)
            self.last_post_time[key] = self.last_post_time.get(key, 0) + 1
        return plan

[account_01]
platform = 平台A
account_id = xxxxx
rtmp_url = rtmp://push.xxx.com/live
stream_key = sk_live_xxx
category = 劳动纠纷
publish_time = 10:00

[account_02]
platform = 平台A
account_id = yyyyy
rtmp_url = rtmp://push.xxx.com/live
stream_key = sk_live_yyy
category = 合同纠纷
publish_time = 14:00

def append_compliance_disclaimer(script: str, format_type: str = "text") -> str:
    """
    为法律内容追加合规声明
    format_type: text / subtitle / voiceover
    """
    disclaimer_text = (
        "本内容仅供法律知识科普参考，不构成任何形式的法律意见或建议。"
        "具体法律问题请咨询具有执业资格的律师。"
    )

    if format_type == "text":
        return script + "\n\n" + disclaimer_text
    elif format_type == "subtitle":
        return disclaimer_text
    elif format_type == "voiceover":
        return "本内容仅供参考，具体问题请咨询专业律师。"
    else:
        return script

from moviepy.editor import VideoFileClip, AudioFileClip, CompositeVideoClip, TextClip, ImageClip

def compose_legal_video(
    audio_path: str,
    avatar_image_path: str,
    subtitle_texts: list,
    disclaimer_text: str,
    output_path: str,
    duration: float
):
    """
    合成法律科普视频，包含三段式字幕和固定免责水印
    subtitle_texts: list of dict，每项包含text、start_time、end_time
    """
    audio = AudioFileClip(audio_path)

    avatar = ImageClip(avatar_image_path).set_duration(duration).resize(height=1080)

    subtitle_clips = []
    for item in subtitle_texts:
        clip = TextClip(
            item["text"],
            fontsize=36,
            color='white',
            stroke_color='black',
            stroke_width=2,
            method='caption',
            size=(avatar.w * 0.85, None)
        ).set_start(item["start_time"]).set_duration(
            item["end_time"] - item["start_time"]
        ).set_position(('center', avatar.h * 0.82))
        subtitle_clips.append(clip)

    disclaimer_clip = TextClip(
        disclaimer_text,
        fontsize=24,
        color='#CCCCCC',
        stroke_color='black',
        stroke_width=1,
        font='NotoSansSC-Regular'
    ).set_duration(duration).set_position(('center', avatar.h * 0.94))

    final = CompositeVideoClip([avatar, *subtitle_clips, disclaimer_clip])
    final = final.set_audio(audio)

    final.write_videofile(
        output_path,
        codec='libx264',
        audio_codec='aac',
        fps=24,
        bitrate='2500k'
    )

legal_pronunciation_map = {
    "诉讼": "su song",
    "仲裁": "zhong cai",
    "劳动合同": "lao dong he tong",
}

def filter_topics_by_category(kb: dict, category: str) -> list:
    return [t for t in kb["topics"] if t["category"] == category]