AI 集成模式折腾手记

AI集成 API调用 嵌入模式 混合架构 Prompt工程
作者: 醉月思 发布日期: 2023-03-17 分类: 技术实践

这次踩了不少坑，也积累了些经验。

最开始也是最简单的模式：用户请求进来，后端调 AI API，拿结果返回。

最初的"简单调用"

最开始也是最简单的模式：用户请求进来，后端调 AI API，拿结果返回。

# 最开始的实现
import openai
from fastapi import FastAPI

app = FastAPI()
openai.api_key = "sk-..."

@app.post("/generate-summary")
async def generate_summary(content: str):
    response = openai.ChatCompletion.create(
        model="gpt-4",
        messages=[
            {"role": "system", "content": "你是个内容摘要专家"},
            {"role": "user", "content": f"请为以下内容生成摘要：\n{content}"}
        ]
    )
    return {"summary": response.choices[0].message.content}

这个实现看起来很简单，但问题很快就暴露出来：

第一，API 调用太慢。 用户提交内容等 10-20 秒才有结果，体验很差。加上网络波动、API 限流，有时候要等半分钟。

第二，成本没控制住。 用 GPT-4 每次调用都要几美分，量大起来成本不可控。后来改成 GPT-3.5，便宜是便宜了，但质量明显下降，生成的摘要有时候抓不住重点。

第三，错误处理不够健壮。 API 限流、网络超时、模型异常，各种情况都有可能。一开始没考虑周全，用户看到的直接是 500 错误。

第四，Prompt 没调好。 同样的内容，有时候生成得很准确，有时候就跑偏。比如产品介绍被理解成新闻，技术文档被理解成营销文案。

这些问题逼着我们重新想：AI 集成得把能力嵌进业务流程，不能停在调个 API 就完事。

改进后的"异步调用模式"

第一个改进是把同步调用改成异步。用户提交内容后立即返回，后台异步调用 AI，完成后再通知用户或更新数据。

# 改进后的异步实现
import openai
from fastapi import FastAPI, BackgroundTasks
from redis import Redis
import json

app = FastAPI()
redis = Redis()
openai.api_key = "sk-..."

def process_with_ai(task_id: str, content: str):
    try:
        response = openai.ChatCompletion.create(
            model="gpt-3.5-turbo",
            messages=[
                {"role": "system", "content": "你是个内容处理专家"},
                {"role": "user", "content": f"处理以下内容：生成摘要（100字以内）、提取3-5个标签、生成SEO标题\n\n{content}"}
            ],
            response_format={"type": "json_object"}
        )
        result = json.loads(response.choices[0].message.content)

        # 存储结果
        redis.hset(f"task:{task_id}", mapping={
            "status": "completed",
            "summary": result.get("summary", ""),
            "tags": json.dumps(result.get("tags", [])),
            "seo_title": result.get("seo_title", "")
        })
    except Exception as e:
        redis.hset(f"task:{task_id}", mapping={
            "status": "failed",
            "error": str(e)
        })

@app.post("/process-content")
async def process_content(content: str, background_tasks: BackgroundTasks):
    task_id = f"task_{int(time.time())}"

    # 初始化任务状态
    redis.hset(f"task:{task_id}", mapping={
        "status": "pending",
        "content": content
    })

    # 后台处理
    background_tasks.add_task(process_with_ai, task_id, content)

    return {"task_id": task_id, "status": "pending"}

@app.get("/task/{task_id}")
async def get_task_status(task_id: str):
    task_data = redis.hgetall(f"task:{task_id}")
    if not task_data:
        return {"error": "Task not found"}

    return {
        "task_id": task_id,
        "status": task_data.get(b"status", "unknown").decode(),
        "result": {
            "summary": task_data.get(b"summary", b"").decode(),
            "tags": json.loads(task_data.get(b"tags", b"[]").decode()),
            "seo_title": task_data.get(b"seo_title", b"").decode()
        } if task_data.get(b"status") == b"completed" else None,
        "error": task_data.get(b"error", b"").decode() if task_data.get(b"status") == b"failed" else None
    }

这个改进解决了体验问题：用户不用干等，可以去做别的事，再回来查看结果。但也带来了新的问题：

第一，任务状态管理变复杂了。 Redis 里存的任务状态要考虑清理机制，不然时间长了会堆积很多过期任务。一开始没注意，几个月后 Redis 内存爆满。

第二，用户需要轮询状态。 前端得定时问后端"好了没"，增加了请求量。后来加了 WebSocket 通知，但复杂度又上去了。

第三，结果缓存没做好。 同样的内容反复处理，浪费 API 调用。后来加了内容指纹缓存，先检查缓存，有就直接返回。

混合模式：不同场景用不同模型

成本和质量的平衡是 AI 集成的关键问题。不是所有场景都需要 GPT-4，也不是所有场景都适合 GPT-3.5。

我们的做法是按场景分级：

# 按场景选择模型
class AIModelSelector:
    def __init__(self):
        self.model_mapping = {
            # 高质量场景
            "high_quality": {
                "model": "gpt-4",
                "max_tokens": 2000,
                "temperature": 0.3
            },
            # 通用场景
            "general": {
                "model": "gpt-3.5-turbo",
                "max_tokens": 1000,
                "temperature": 0.7
            },
            # 快速场景
            "fast": {
                "model": "gpt-3.5-turbo-16k",
                "max_tokens": 500,
                "temperature": 0.5
            }
        }

    def select_model(self, task_type: str, priority: str = "normal"):
        # 高优先级任务用高质量模型
        if priority == "high":
            return self.model_mapping["high_quality"]

        # 根据任务类型选择
        if task_type in ["content_creation", "code_review", "complex_analysis"]:
            return self.model_mapping["high_quality"]
        elif task_type in ["summarization", "tag_extraction", "title_generation"]:
            return self.model_mapping["general"]
        else:
            return self.model_mapping["fast"]

# 使用示例
selector = AIModelSelector()

def generate_summary(content: str, priority: str = "normal"):
    model_config = selector.select_model("summarization", priority)

    response = openai.ChatCompletion.create(
        model=model_config["model"],
        max_tokens=model_config["max_tokens"],
        temperature=model_config["temperature"],
        messages=[...]
    )

    return response.choices[0].message.content

这个分级策略把 API 成本控制下来了：GPT-4 只占 10% 的调用量，但处理的是最重要的任务。GPT-3.5 处理 70%，还有 20% 的简单任务用更便宜的模型。

但分级也有坑：

第一，分级标准不好定。 什么任务算"高质量"，什么算"快速"，边界有时候很模糊。一开始定得太死，后来加了人工审核和自动调整机制。

第二，用户不理解。 同样的功能，有时候快有时候慢，用户会困惑。后来在 UI 上加了优先级选项，让用户自己选。

第三，监控和分析不够。 不知道各个模型的实际效果如何。后来加了详细的日志和监控，定期分析调用情况和用户反馈。

嵌入模式：AI 成为系统的一部分

再往下走，集成重点是把 AI 嵌进工作流——参与决策、生成内容、改流程，而不是单独挂一个接口。

举个例子：内容审核流程。

# 嵌入式 AI 审核流程
class ContentAIWorkflow:
    def __init__(self):
        self.ai_client = openai
        self.db = Database()
        self.notification = NotificationService()

    async def process_content(self, content_id: int, content: dict):
        # 第一阶段：AI 初步评估
        ai_assessment = await self._ai_pre_assessment(content)

        # 根据评估结果走不同流程
        if ai_assessment["risk_level"] == "high":
            # 高风险内容，转人工审核
            await self._escalate_to_human(content_id, ai_assessment)
        elif ai_assessment["risk_level"] == "medium":
            # 中风险，补充信息后重新评估
            additional_checks = await self._run_additional_checks(content, ai_assessment)
            if additional_checks["passed"]:
                await self._approve_content(content_id)
            else:
                await self._escalate_to_human(content_id, additional_checks)
        else:
            # 低风险，直接通过
            await self._approve_content(content_id)

        # AI 学习阶段：记录反馈，持续优化
        await self._record_feedback(content_id, ai_assessment)

    async def _ai_pre_assessment(self, content: dict):
        prompt = f"""
        请评估以下内容的风险等级，返回 JSON 格式：

        内容类型：{content['type']}
        内容标题：{content['title']}
        内容正文：{content['body'][:500]}...

        评估维度：
        1. 敏感词风险
        2. 虚假信息风险
        3. 版权风险
        4. 合规性风险

        返回格式：
        {{
            "risk_level": "high|medium|low",
            "risk_factors": ["敏感词", "虚假信息"],
            "confidence": 0.85,
            "suggestion": "建议人工审核"
        }}
        """

        response = await self.ai_client.ChatCompletion.acreate(
            model="gpt-4",
            messages=[
                {"role": "system", "content": "你是内容审核专家"},
                {"role": "user", "content": prompt}
            ],
            response_format={"type": "json_object"}
        )

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

    async def _escalate_to_human(self, content_id: int, assessment: dict):
        # 标记为待人工审核
        await self.db.update(
            "contents",
            {"id": content_id},
            {"status": "pending_review", "ai_assessment": assessment}
        )

        # 通知审核人员
        await self.notification.send_to_reviewers({
            "content_id": content_id,
            "risk_level": assessment["risk_level"],
            "factors": assessment["risk_factors"]
        })

    async def _record_feedback(self, content_id: int, assessment: dict):
        # 记录 AI 评估结果和最终决策，用于后续优化
        final_decision = await self.db.get_one("contents", {"id": content_id})

        await self.db.insert("ai_feedback", {
            "content_id": content_id,
            "ai_prediction": assessment["risk_level"],
            "ai_confidence": assessment["confidence"],
            "final_decision": final_decision["status"],
            "timestamp": datetime.now()
        })

这个嵌入模式和简单调用的区别：

AI 参与决策，不只吐一段生成内容。 它给出评估和建议，系统再按风险等级分流到不同业务流程。

AI 有反馈循环。 记录 AI 的预测和最终决策，定期分析准确率，调整 Prompt 和参数。

AI 是"专家"不是"工人"。 AI 用来做它擅长的事（理解语义、识别模式），复杂的业务规则、合规检查还是交给规则引擎和人工。

这种模式的坑也不少：

第一，边界情况太多。 AI 评估有时候会出错，尤其是边界情况。比如幽默内容被误判为违规，专业术语被误判为敏感词。加了大量的示例和负样本，但还是难免有漏网。

第二，人机协作流程复杂。 什么时候该让 AI 做，什么时候该让人工做，界限不好划。一开始 AI 太激进，后来加了人工复核的优先队列。

第三，AI 模型更新影响大。 模型一更新，评估结果可能变化，导致审核标准不一致。后来加了模型版本管理和 A/B 测试。

混合架构：规则 + AI + 人工

最后我们采用的是混合架构：规则引擎处理明确的规则，AI 处理模糊的判断，人工处理复杂的边界。

graph TD A[内容提交] --> B{规则引擎检查} B -->|明确违规| C[直接拒绝] B -->|明确通过| D[通过审核] B -->|不明确| E{AI 评估} E -->|高风险| F[人工审核] E -->|低风险| D E -->|中风险| G{补充检查} G -->|通过| D G -->|不通过| F F --> H{人工决策} H -->|通过| D H -->|拒绝| C H -->|修改| I[返回修改] D --> J[记录反馈] C --> J I --> A J --> K[持续优化]

这个架构的优势：

明确规则优先。 比如明确的敏感词、明显的垃圾广告，规则引擎直接处理，又快又准。

AI 处理复杂判断。 比如内容是否涉及虚假信息、是否有版权风险，这些靠规则很难判断，但 AI 能理解上下文。

人工处理边界。 比如幽默内容的边界、灰色地带的判断，最终还是需要人工拍板。

持续优化。 三个环节都有反馈记录，定期分析，优化规则、Prompt 和人工审核标准。

这个架构的坑：

第一，三个环节的协调问题。 规则引擎改了，AI 的 Prompt 也要改，人工审核标准也要调。三个环节容易不一致。

第二，性能和成本的平衡。 规则引擎很快，AI 很慢，人工审核最慢。要在准确率和效率之间找平衡。

第三，复杂度明显增加。 三个环节要各自监控、各自优化，还要整体协调。系统复杂度上来后，维护成本也高了。

折腾大半年留下的几条

这次 AI 集成折腾了大半年，几条硬经验：

第一，不要把 AI 当万能药。 AI 有擅长的地方，也有不擅长的地方。用它来做理解、生成、分类这些事，用它来做规则判断、精确计算，那就是找错了地方。

第二，从简单开始，逐步复杂化。 一开始就是复杂的混合架构，很容易把自己绕进去。先从简单的 API 调用开始，遇到问题再逐步优化。

第三，监控和日志要到位。 AI 的行为不像传统代码那么可预测，没有详细的监控和日志，出问题很难排查。

第四，要有降级方案。 API 挂了、限流了、模型异常了，系统要有降级方案，不能因为 AI 出问题就把整个系统拖垮。

第五，成本控制要前置。 不要等账单来了才发现成本爆了。从设计阶段就要考虑成本控制，按场景分级、加缓存、用更便宜的模型。

第六，AI 需要调优。 Prompt、参数、模型选择，这些都需要持续调优。不要指望一次就完美，把它当作一个持续优化的过程。

AI 集成没有终点，从同步调 API 到嵌入工作流、再到规则 + AI + 人工，每一层都是踩坑换来的。接好了，AI 是系统里的一环；接不好，就是个慢且贵的外挂。

这次改造从简单 API 调用切到混合架构后，API 成本降了 60%，审核准确率从 75% 提到 92%，人工审核量少了 80%。

版权声明: 本文首发于 指尖魔法屋-AI 集成模式折腾手记（https://blog.thinkmoon.cn/post/198-ai-integration-patterns-embedding-practice/） 转载或引用必须申明原指尖魔法屋来源及源地址！

相关阅读

根据标签、分类和主题为你挑选的相近文章