11 - 实际应用案例¶

from fastapi import FastAPI, HTTPException
from pydantic import BaseModel
import torch
from transformers import AutoModelForCausalLM, AutoTokenizer

app = FastAPI(title="智能客服系统")

# 加载模型
# 注意：正确的模型ID为 deepseek-ai/DeepSeek-R1（区分大小写）
# 可选蒸馏版本：deepseek-ai/DeepSeek-R1-Distill-Qwen-1.5B 等
model_name = "deepseek-ai/DeepSeek-R1"
tokenizer = AutoTokenizer.from_pretrained(model_name)
model = AutoModelForCausalLM.from_pretrained(
    model_name,
    torch_dtype=torch.float16,
    device_map="auto"
)

class CustomerQuery(BaseModel):  # BaseModel Pydantic数据验证模型
    user_id: str
    query: str
    conversation_history: list[dict] | None = None

class CustomerResponse(BaseModel):
    response: str
    confidence: float
    suggested_actions: list[str]

class CustomerServiceSystem:
    """
    智能客服系统
    """
    def __init__(self, model, tokenizer):
        self.model = model
        self.tokenizer = tokenizer
        self.knowledge_base = self._load_knowledge_base()

    def _load_knowledge_base(self):
        """
        加载知识库
        """
        # 这里简化处理，实际应该从数据库或文件加载
        return {
            "产品信息": "我们的产品包括...",
            "价格信息": "产品价格如下...",
            "售后服务": "我们提供7x24小时客服...",
            "退换货政策": "退换货政策如下..."
        }

    def _search_knowledge_base(self, query: str) -> str:
        """
        搜索知识库
        """
        # 简化的搜索逻辑
        for topic, content in self.knowledge_base.items():
            if topic in query:
                return content

        return ""

    def _build_prompt(self, query: str, history: list[dict] = None) -> str:
        """
        构建提示词
        """
        prompt = "你是一个专业的客服代表。请根据以下信息回答用户问题。\n\n"

        # 添加知识库信息
        knowledge = self._search_knowledge_base(query)
        if knowledge:
            prompt += f"相关知识：\n{knowledge}\n\n"

        # 添加对话历史
        if history:
            prompt += "对话历史：\n"
            for msg in history[-5:]:  # 只保留最近5轮对话
                role = msg.get("role", "user")
                content = msg.get("content", "")
                prompt += f"{role}: {content}\n"
            prompt += "\n"

        # 添加当前问题
        prompt += f"用户问题：{query}\n\n请提供准确、友好、专业的回答："

        return prompt

    def generate_response(self, query: CustomerQuery) -> CustomerResponse:
        """
        生成响应
        """
        # 构建提示词
        prompt = self._build_prompt(
            query.query,
            query.conversation_history
        )

        # 编码输入
        inputs = self.tokenizer(prompt, return_tensors="pt").to(self.model.device)

        # 生成响应
        with torch.no_grad():  # 禁用梯度计算，节省内存
            outputs = self.model.generate(
                **inputs,
                max_length=300,
                temperature=0.7,
                top_p=0.9,
                do_sample=True
            )

        # 解码响应
        response = self.tokenizer.decode(outputs[0], skip_special_tokens=True)

        # 提取客服回答
        if "请提供" in response:
            response = response.split("请提供")[-1].strip()  # [-1]负索引取最后元素

        # 计算置信度（简化）
        confidence = 0.85  # 实际应该基于模型输出计算

        # 生成建议操作
        suggested_actions = self._generate_suggested_actions(query.query, response)

        return CustomerResponse(
            response=response,
            confidence=confidence,
            suggested_actions=suggested_actions
        )

    def _generate_suggested_actions(self, query: str, response: str) -> list[str]:
        """
        生成建议操作
        """
        actions = []

        # 基于查询类型生成建议
        if "价格" in query or "多少钱" in query:
            actions.append("提供详细价格表")
            actions.append("推荐优惠活动")

        if "退货" in query or "退款" in query:
            actions.append("查看退换货政策")
            actions.append("联系人工客服")

        if "产品" in query:
            actions.append("推荐相关产品")
            actions.append("提供产品详情")

        if not actions:
            actions.append("记录用户反馈")
            actions.append("提供进一步帮助")

        return actions

# 全局客服系统
customer_service = CustomerServiceSystem(model, tokenizer)

@app.post("/customer-service", response_model=CustomerResponse)
async def handle_customer_query(query: CustomerQuery):  # async定义异步函数
    """
    处理客户查询
    """
    try:  # try/except捕获异常
        response = customer_service.generate_response(query)
        return response
    except Exception as e:
        raise HTTPException(status_code=500, detail=str(e))

if __name__ == "__main__":
    import uvicorn
    uvicorn.run(app, host="0.0.0.0", port=8000)

class CodeGenerationSystem:
    """
    代码生成系统
    """
    def __init__(self, model, tokenizer):
        self.model = model
        self.tokenizer = tokenizer

    def generate_code(self, description: str, language: str = "python") -> str:
        """
        生成代码

        Args:
            description: 代码描述
            language: 编程语言
        """
        # 构建提示词
        prompt = f"""请根据以下描述生成{language}代码：

描述：{description}

要求：
1. 代码应该清晰、高效
2. 添加必要的注释
3. 遵循最佳实践
4. 包含错误处理

代码："""

        # 编码输入
        inputs = self.tokenizer(prompt, return_tensors="pt").to(self.model.device)

        # 生成代码
        with torch.no_grad():
            outputs = self.model.generate(
                **inputs,
                max_length=500,
                temperature=0.3,  # 代码生成使用较低温度
                top_p=0.95,
                do_sample=True
            )

        # 解码代码
        code = self.tokenizer.decode(outputs[0], skip_special_tokens=True)

        # 提取代码部分
        if "代码：" in code:
            code = code.split("代码：")[-1].strip()

        return code

    def explain_code(self, code: str) -> str:
        """
        解释代码

        Args:
            code: 要解释的代码
        """
        # 构建提示词
        prompt = f"""请详细解释以下代码的功能和工作原理：

<code>
{code}
</code>

请包括：
1. 代码的整体功能
2. 关键部分的解释
3. 可能的改进建议

解释："""

        # 编码输入
        inputs = self.tokenizer(prompt, return_tensors="pt").to(self.model.device)

        # 生成解释
        with torch.no_grad():
            outputs = self.model.generate(
                **inputs,
                max_length=400,
                temperature=0.5,
                top_p=0.9,
                do_sample=True
            )

        # 解码解释
        explanation = self.tokenizer.decode(outputs[0], skip_special_tokens=True)

        # 提取解释部分
        if "解释：" in explanation:
            explanation = explanation.split("解释：")[-1].strip()

        return explanation

# 使用示例
# code_system = CodeGenerationSystem(model, tokenizer)
# code = code_system.generate_code("实现一个快速排序算法")
# explanation = code_system.explain_code(code)

class ContentCreationSystem:
    """
    内容创作系统
    """
    def __init__(self, model, tokenizer):
        self.model = model
        self.tokenizer = tokenizer

    def generate_article(self, topic: str, style: str = "专业") -> str:
        """
        生成文章

        Args:
            topic: 文章主题
            style: 写作风格
        """
        # 构建提示词
        prompt = f"""请写一篇关于"{topic}"的文章。

风格要求：{style}

文章结构：
1. 引人入胜的开头
2. 清晰的主体内容
3. 有力的结尾

文章："""

        # 编码输入
        inputs = self.tokenizer(prompt, return_tensors="pt").to(self.model.device)

        # 生成文章
        with torch.no_grad():
            outputs = self.model.generate(
                **inputs,
                max_length=800,
                temperature=0.8,  # 创作使用较高温度
                top_p=0.95,
                do_sample=True
            )

        # 解码文章
        article = self.tokenizer.decode(outputs[0], skip_special_tokens=True)

        # 提取文章部分
        if "文章：" in article:
            article = article.split("文章：")[-1].strip()

        return article

    def generate_social_media_post(self, topic: str, platform: str = "微博") -> str:
        """
        生成社交媒体帖子

        Args:
            topic: 帖子主题
            platform: 平台类型
        """
        # 根据平台调整要求
        if platform == "微博":
            max_length = 140
            style = "简洁、有吸引力"
        elif platform == "微信公众号":
            max_length = 2000
            style = "详细、有价值"
        else:
            max_length = 280
            style = "简洁、有趣"

        # 构建提示词
        prompt = f"""请为{platform}写一个关于"{topic}"的帖子。

要求：
1. 风格：{style}
2. 长度：不超过{max_length}字
3. 包含相关标签
4. 引人互动

帖子："""

        # 编码输入
        inputs = self.tokenizer(prompt, return_tensors="pt").to(self.model.device)

        # 生成帖子
        with torch.no_grad():
            outputs = self.model.generate(
                **inputs,
                max_length=max_length + 100,
                temperature=0.9,
                top_p=0.95,
                do_sample=True
            )

        # 解码帖子
        post = self.tokenizer.decode(outputs[0], skip_special_tokens=True)

        # 提取帖子部分
        if "帖子：" in post:
            post = post.split("帖子：")[-1].strip()

        # 截断到最大长度
        post = post[:max_length]

        return post

# 使用示例
# content_system = ContentCreationSystem(model, tokenizer)
# article = content_system.generate_article("人工智能的发展趋势", "科普")
# post = content_system.generate_social_media_post("新产品发布", "微博")

# 不好的提示词
bad_prompt = "写一个排序算法"

# 好的提示词
good_prompt = """请用Python实现一个快速排序算法。

要求：
1. 时间复杂度：O(n log n)
2. 空间复杂度：O(log n)
3. 包含详细的注释
4. 添加错误处理
5. 提供测试用例

示例：
输入：[3, 1, 4, 1, 5, 9, 2, 6]
输出：[1, 1, 2, 3, 4, 5, 6, 9]

请实现："""

class OptimizedInferenceService:
    """
    优化的推理服务
    """
    def __init__(self, model, tokenizer):
        self.model = model
        self.tokenizer = tokenizer
        self.cache = {}

    def batch_generate(self, prompts: list[str], batch_size=8):
        """
        批量生成
        """
        results = []

        for i in range(0, len(prompts), batch_size):
            batch = prompts[i:i+batch_size]

            # 编码批次
            inputs = self.tokenizer(
                batch,
                padding=True,
                truncation=True,
                return_tensors="pt"
            ).to(self.model.device)

            # 批量生成
            with torch.no_grad():
                outputs = self.model.generate(
                    **inputs,
                    max_length=200,
                    temperature=0.7,
                    do_sample=True
                )

            # 解码结果
            for output in outputs:
                result = self.tokenizer.decode(output, skip_special_tokens=True)
                results.append(result)

        return results

    def cached_generate(self, prompt: str, use_cache=True):
        """
        带缓存的生成
        """
        # 检查缓存
        if use_cache and prompt in self.cache:
            return self.cache[prompt]

        # 生成结果
        inputs = self.tokenizer(prompt, return_tensors="pt").to(self.model.device)

        with torch.no_grad():
            outputs = self.model.generate(
                **inputs,
                max_length=200,
                temperature=0.7,
                do_sample=True
            )

        result = self.tokenizer.decode(outputs[0], skip_special_tokens=True)

        # 缓存结果
        if use_cache:
            self.cache[prompt] = result

        return result

# 使用示例
# service = OptimizedInferenceService(model, tokenizer)
# results = service.batch_generate(prompts, batch_size=8)
# result = service.cached_generate("什么是机器学习？")

import logging

class RobustInferenceService:
    """
    健壮的推理服务
    """
    def __init__(self, model, tokenizer):
        self.model = model
        self.tokenizer = tokenizer
        self.logger = logging.getLogger(__name__)

    def safe_generate(self, prompt: str, max_retries=3):
        """
        安全的生成（带重试）
        """
        for attempt in range(max_retries):
            try:
                # 验证输入
                if not prompt or not isinstance(prompt, str):  # isinstance检查类型
                    raise ValueError("Invalid prompt")

                if len(prompt) > 10000:
                    raise ValueError("Prompt too long")

                # 生成结果
                inputs = self.tokenizer(prompt, return_tensors="pt").to(self.model.device)

                with torch.no_grad():
                    outputs = self.model.generate(
                        **inputs,
                        max_length=200,
                        temperature=0.7,
                        do_sample=True
                    )

                result = self.tokenizer.decode(outputs[0], skip_special_tokens=True)

                # 验证输出
                if not result:
                    raise ValueError("Empty result")

                return result

            except ValueError as e:
                self.logger.warning(f"Validation error (attempt {attempt+1}): {e}")
                if attempt == max_retries - 1:
                    # 返回默认响应
                    return "抱歉，我无法处理您的请求。请稍后再试。"

            except Exception as e:
                self.logger.error(f"Generation error (attempt {attempt+1}): {e}")
                if attempt == max_retries - 1:
                    # 返回错误响应
                    return "抱歉，发生了错误。请稍后再试。"

        return None

# 使用示例
# service = RobustInferenceService(model, tokenizer)
# result = service.safe_generate("什么是机器学习？")

# 使用思维链
cot_prompt = """让我们一步步思考：

问题：{question}

步骤1：分析问题
步骤2：确定关键信息
步骤3：应用相关知识
步骤4：得出结论

答案："""

# 使用少样本学习
few_shot_prompt = """例子1：
输入：2 + 2 = ?
输出：4

例子2：
输入：3 + 3 = ?
输出：6

问题：{question}
答案："""

# 根据任务调整参数
task_params = {
    "code_generation": {
        "temperature": 0.3,
        "top_p": 0.95,
        "max_length": 500
    },
    "creative_writing": {
        "temperature": 0.9,
        "top_p": 0.95,
        "max_length": 1000
    },
    "qa": {
        "temperature": 0.5,
        "top_p": 0.9,
        "max_length": 200
    }
}

# 使用示例
# params = task_params["code_generation"]
# outputs = model.generate(input_ids, **params)

import re

def post_process_output(output: str, task_type: str) -> str:
    """
    后处理输出
    """
    # 移除重复内容
    output = re.sub(r'(.)\1{3,}', r'\1\1\1', output)

    # 根据任务类型处理
    if task_type == "code":
        # 提取代码块
        code_match = re.search(r'```python\n(.*?)\n```', output, re.DOTALL)  # re.search在字符串中搜索匹配模式
        if code_match:
            output = code_match.group(1)

    elif task_type == "qa":
        # 提取答案部分
        if "答案：" in output:
            output = output.split("答案：")[-1].strip()

    # 移除多余空行
    output = re.sub(r'\n{3,}', '\n\n', output)

    return output

# 使用示例
# processed_output = post_process_output(raw_output, "qa")

# 练习: 实现一个简单的智能客服系统
class CustomerService:
    def __init__(self, model, tokenizer):
        # 你的代码
        pass

    def handle_query(self, query):
        # 你的代码
        pass

# 练习: 实现一个简单的代码生成系统
class CodeGenerator:
    def __init__(self, model, tokenizer):
        # 你的代码
        pass

    def generate(self, description):
        # 你的代码
        pass

# 练习: 实现一个内容创作系统
class ContentCreator:
    def __init__(self, model, tokenizer):
        # 你的代码
        pass

    def create_article(self, topic, style):
        # 你的代码
        pass

    def create_social_post(self, topic, platform):
        # 你的代码
        pass

# 练习: 实现一个优化的推理服务
class OptimizedService:
    def __init__(self, model, tokenizer):
        # 你的代码
        pass

    def batch_generate(self, prompts):
        # 你的代码
        pass

    def cached_generate(self, prompt):
        # 你的代码
        pass