AI SaaS 开发实战

六个项目，一条线

这 6 个项目不是孤立的，它们共享同一套基础设施：FastAPI + SQLAlchemy 2.0(async) + MySQL + Redis。从 Basic_Web 脚手架开始，往上搭业务。

FinBuddy_Web — LLM 代理服务

FinBuddy_Web 是 FinBuddy 的云端 SaaS 后端，最核心的是 LLM 代理服务。三层控制：层级限额检查 + 余额预检 + Redis 分布式信号量并发控制。

# LLM 代理 — 层级限额 + 余额预检 + 并发控制
class ProxyService:
    async def chat(self, req: ChatRequest, user: User):
        # 1. 层级限额：tier2/tier3 模型有每日使用次数限制
        await self._check_tier_limit(user, req.model)

        # 2. 余额预检：调用前检查余额是否 > 0
        balance = await self.billing.get_balance(user.id)
        if balance <= 0:
            raise InsufficientBalanceError()

        # 3. 并发控制：Redis 分布式信号量
        async with self.semaphore(user.id):
            async for chunk in self._stream(req):
                yield chunk

流式响应用独立短生命周期 Session，避免长时间占用连接池导致 QueuePool 耗尽。这个坑我踩过——高并发时整个服务卡死。FinBuddy_Web 的 LLM 代理也用在投资场景——帮用户调取财报数据、做量化筛选，所以并发控制特别重要。

Promo_Web — Skill 组合引擎

Promo 的核心创新是把营销内容生成抽象为可组合的"技能链"：framework → style → adapter → enhancer，四层 Skill 按类型顺序拼接 Prompt。

# Skill 四层组合机制
class GenerationEngine:
    def assemble_prompt(self, product, persona, channel,
                        framework_skill, style_skill,
                        adapter_skill, enhancer_skills=None):
        variables = {
            "product_name": product.name,
            "selling_points": "、".join(product.selling_points),
            "brand_tone": product.brand_tone or "专业、可信赖",
            "persona_desc": f"{persona.description}。"
                           f"语言风格：{persona.language_style}",
            "channel_rules": self._format_channel_rules(channel),
        }
        parts = [
            framework_skill.render(variables),
            style_skill.render(variables),
            adapter_skill.render(variables),
        ]
        for enhancer in (enhancer_skills or []):
            parts.append(enhancer.render(variables))
        return "\n\n".join(parts)

SkillStore — 技能自动分类

SkillStore 用正则规则引擎，根据技能名称和描述中的中英文关键词自动分类到 domain/category 二级分类体系。

# 技能自动分类引擎
_CLASSIFICATION_RULES = [
    (r"orchestrat|编排|流水线|pipeline", "orchestrate", "orchestrator"),
    (r"market|营销|推广|growth|转化", "content", "marketing"),
    (r"writ|写作|文案|article|copy", "content", "writing"),
    (r"video|remotion|视频|媒体|image", "tool", "media"),
    (r"design|设计|hig|platform|ios|macos", "develop", "design"),
]

def _classify_skill(name: str, description: str) -> tuple:
    text = f"{name} {description}".lower()
    for pattern, domain, category in _CLASSIFICATION_RULES:
        if re.search(pattern, text):
            return domain, category
    return "develop", "dev-assist"

MarketingVideo — Remotion 视频管线

用 Remotion 4 + Python 管线做营销视频自动生成。4 步流水线：TTS 配音 -> Whisper 字幕 -> Remotion 渲染 -> 平台发布。

# Python 4 步管线
class VideoPipeline:
    async def run(self, job: VideoJob):
        # Step 1: TTS 配音
        audio = await self.tts.generate(job.script, job.voice)
        # Step 2: Whisper 字幕
        subtitles = await self.whisper.transcribe(audio)
        # Step 3: Remotion 渲染
        video = await self.renderer.render(job.template, subtitles)
        # Step 4: 平台发布
        await self.publisher.publish(video, job.platform)

最头疼的是 Whisper 中文识别金融术语经常出错，需要加自定义词典后处理。TTS 引擎做了路由：默认 Edge TTS（免费），重要内容用 OpenAI TTS（音质好但贵）。

Basic_Web — SaaS 脚手架

所有项目的基础。最值得说的是分布式信号量——基于 Redis INCR + EXPIRE 实现跨实例并发控制。

# 分布式信号量 — 跨实例并发控制
class DistributedSemaphore:
    async def acquire(self) -> bool:
        current = await redis_client.incr(self._key)
        if current == 1:
            await redis_client.expire(self._key, self._ttl)
        if current > self._max_concurrent:
            await redis_client.decr(self._key)
            return False
        return True