电影解说AI Agent实战项目

目录

1. 项目概述
2. 系统架构设计
3. 核心组件实现
4. Prompt设计
5. 结构化输出设计
6. 完整实现代码
7. 使用示例
8. 项目优化建议
9. 总结与最佳实践

---

1. 项目概述

1.1 项目目标

本项目构建一个电影解说AI Agent，能够：

• 接收电影名称或相关信息
• 基于LLM知识库回答电影相关问题
• 生成专业的电影解说文案
• 支持结构化输出（包含评分、推荐、关键要点等）
• 支持多轮对话，记住对话历史

1.2 技术栈

核心技术组件：

• LangChain: 应用框架，提供LLM调用、记忆管理等功能
• ChatOpenAI: LLM接口，支持OpenAI和DeepSeek API
• Pydantic: 数据验证和结构化输出
• ConversationBufferMemory: 对话记忆管理

1.3 项目特点

1. 结构化输出：使用Pydantic模型定义输出格式，包含评分、推荐、关键要点等结构化信息
2. 双模式支持：支持结构化输出和简单文本输出两种模式
3. 多API支持：同时支持OpenAI和DeepSeek API
4. 对话记忆：支持多轮对话，记住历史上下文
5. 智能解析：强大的JSON解析器，能够处理各种格式的LLM输出

---

2. 系统架构设计

2.1 架构流程图

2.2 核心组件

组件	功能	实现方式
Agent核心	任务编排和LLM调用	SimpleAgent类
LLM引擎	理解和生成	ChatOpenAI (OpenAI/DeepSeek)
输出解析器	解析和格式化输出	MovieOutputParser
记忆系统	保存对话历史	ConversationBufferMemory
Prompt模板	指导生成风格	ChatPromptTemplate
数据模型	结构化输出定义	MovieResponse (Pydantic)

2.3 数据流

---

3. 核心组件实现

3.1 结构化输出模型

使用Pydantic定义电影解说的结构化输出格式：

from typing import List, Optional, Literal
from pydantic import BaseModel, Field

class MovieResponse(BaseModel):
    """电影解说响应格式"""
    
    # 主要回答内容（必需）
    answer: str = Field(
        description="对用户问题的详细回答，应该完整、准确、友好、生动"
    )
    
    # 电影名称（可选）
    movie_name: Optional[str] = Field(
        default=None,
        description="如果问题涉及特定电影，提供电影名称"
    )
    
    # 回答分类（必需，有默认值）
    category: Literal[
        "电影介绍",
        "剧情分析",
        "角色分析",
        "主题探讨",
        "技术评价",
        "推荐建议",
        "比较分析",
        "其他"
    ] = Field(
        default="其他",
        description="回答所属的分类"
    )
    
    # 关键要点列表（可选）
    key_points: Optional[List[str]] = Field(
        default=None,
        description="回答中的关键要点列表，用于快速了解核心内容"
    )
    
    # 评分（可选，0-10分）
    rating: Optional[float] = Field(
        default=None,
        ge=0.0,
        le=10.0,
        description="如果涉及电影评分，提供评分（0-10分）"
    )
    
    # 推荐标识（可选）
    recommendation: Optional[bool] = Field(
        default=None,
        description="是否推荐观看这部电影"
    )
    
    # 相关电影推荐（可选）
    related_movies: Optional[List[str]] = Field(
        default=None,
        description="相关的其他电影推荐列表"
    )

设计要点：

• answer字段是必需的，包含主要回答内容
• 其他字段都是可选的，根据问题内容动态填充
• 使用Literal类型限制分类值，确保数据一致性
• 使用Field添加描述，帮助LLM理解字段含义

3.2 输出解析器

MovieOutputParser类负责解析LLM的输出：

from langchain_core.output_parsers import PydanticOutputParser
import json
import re

class MovieOutputParser:
    """电影解说系统输出解析器"""
    
    def __init__(self):
        """初始化解析器"""
        # 使用LangChain的PydanticOutputParser
        self.parser = PydanticOutputParser(pydantic_object=MovieResponse)
        self.format_instructions = self.parser.get_format_instructions()
    
    def parse(self, text: str) -> MovieResponse:
        """
        解析LLM输出
        
        解析策略（按优先级）：
        1. 尝试提取JSON并解析
        2. 使用Pydantic parser解析
        3. 如果都失败，创建默认响应（使用原始文本作为answer）
        """
        if not text or not text.strip():
            return self._create_default_response()
        
        # 策略1: 提取JSON
        json_str = self._extract_json(text)
        if json_str:
            try:
                data = json.loads(json_str)
                if isinstance(data, dict) and "answer" in data:
                    try:
                        return MovieResponse(**data)
                    except Exception:
                        # 字段验证失败，使用answer字段创建响应
                        return self._create_response_from_dict(data)
            except (json.JSONDecodeError, ValueError, TypeError):
                pass
        
        # 策略2: 使用Pydantic parser
        try:
            return self.parser.parse(text)
        except Exception:
            pass
        
        # 策略3: 默认响应
        return MovieResponse(
            answer=text,
            movie_name=None,
            category="其他",
            key_points=None,
            rating=None,
            recommendation=None,
            related_movies=None
        )
    
    def _extract_json(self, text: str) -> Optional[str]:
        """
        从文本中提取JSON内容
        
        支持多种格式：
        - 代码块中的JSON: ```json {...} ```
        - 纯JSON对象: {...}
        - 包含answer字段的JSON
        """
        text = text.strip()
        
        # 尝试提取代码块中的JSON
        json_patterns = [
            r"```json\s*(\{.*?\})\s*```",
            r"```\s*(\{.*?\})\s*```",
        ]
        
        for pattern in json_patterns:
            match = re.search(pattern, text, re.DOTALL)
            if match:
                json_str = match.group(1).strip()
                if self._is_valid_json_data(json_str):
                    return json_str
        
        # 尝试直接提取JSON对象
        if text.startswith("{"):
            last_brace = text.rfind("}")
            if last_brace > 0:
                json_str = text[:last_brace + 1]
                if self._is_valid_json_data(json_str):
                    return json_str
        
        return None
    
    def format_response(self, response: MovieResponse, verbose: bool = True) -> str:
        """
        格式化响应为友好的文本格式
        
        Args:
            response: MovieResponse对象
            verbose: 是否显示详细信息（关键要点、评分等）
        
        Returns:
            格式化后的文本
        """
        output_parts = [response.answer]
        
        if verbose:
            # 关键要点
            if response.key_points:
                output_parts.append("\n\n【关键要点】")
                for i, point in enumerate(response.key_points, 1):
                    output_parts.append(f"{i}. {point}")
            
            # 评分
            if response.rating is not None:
                output_parts.append(f"\n【评分】{response.rating}/10")
            
            # 推荐
            if response.recommendation is not None:
                rec_text = "强烈推荐" if response.recommendation else "不推荐"
                output_parts.append(f"【推荐】{rec_text}")
            
            # 相关电影
            if response.related_movies:
                output_parts.append("\n【相关电影推荐】")
                for movie in response.related_movies:
                    output_parts.append(f"• {movie}")
        
        return "\n".join(output_parts)

设计要点：

• 多策略解析：支持多种JSON格式，提高解析成功率
• 容错处理：即使解析失败也能返回默认响应，保证系统稳定性
• 格式化输出：将结构化数据转换为用户友好的文本格式

---

4. Prompt设计

4.1 系统Prompt

MOVIE_COMMENTATOR_SYSTEM_PROMPT = """
你是一位专业的电影解说员，具有以下特点：

1. **专业素养**
   - 对电影艺术有深入理解
   - 熟悉各种电影类型和风格
   - 能够分析电影的深层含义

2. **解说风格**
   - 语言生动有趣，引人入胜
   - 结构清晰，逻辑严密
   - 既有专业分析，又通俗易懂

3. **内容要求**
   - 介绍电影基本信息（导演、演员、类型等）
   - 概述剧情（避免剧透关键情节）
   - 分析电影主题和艺术特色
   - 评价电影的艺术价值和观赏价值

4. **交互方式**
   - 回答用户关于电影的各种问题
   - 基于对话历史提供连贯的回答
   - 运用你的电影知识直接回答用户的问题

请始终以专业、友好、生动的语调与用户交流。
"""

Prompt设计原则：

1. 角色定位明确：明确Agent是专业电影解说员
2. 能力描述清晰：列出专业素养、风格、内容要求
3. 交互指导：说明如何与用户交互
4. 语气要求：强调专业、友好、生动的语调

4.2 格式说明Prompt

当使用结构化输出模式时，需要添加格式说明：

def get_format_instructions(self) -> str:
    """获取格式说明，用于添加到prompt中"""
    return f"""请严格按照以下JSON格式输出你的回答（只输出JSON数据，不要输出Schema定义）：

{self.format_instructions}

输出示例（这是你需要输出的格式）：
{{
    "answer": "《肖申克的救赎》是一部1994年上映的经典剧情片，由弗兰克·德拉邦特执导...",
    "movie_name": "肖申克的救赎",
    "category": "电影介绍",
    "key_points": [
        "1994年上映的经典剧情片",
        "导演：弗兰克·德拉邦特",
        "主题：希望、自由、友谊、救赎",
        "豆瓣评分9.7分"
    ],
    "rating": 9.7,
    "recommendation": true,
    "related_movies": ["阿甘正传", "当幸福来敲门", "美丽人生"]
}}

重要提示：
1. **只输出JSON数据对象，不要输出Schema定义或格式说明**
2. answer字段是必需的，必须包含完整的回答内容
3. category必须是预定义的分类之一
4. 如果问题不涉及特定电影，movie_name可以为null
5. key_points、rating、recommendation、related_movies可以为空或null
6. 用中文回答，语言要专业、友好、生动、引人入胜
7. 评分范围是0-10分，保留一位小数
8. **直接输出JSON，不要包含任何其他文字说明**"""

格式说明要点：

• 明确格式要求：强调只输出JSON，不要Schema
• 提供示例：给出具体的输出示例
• 重要提示：列出关键注意事项
• 字段说明：说明每个字段的要求和可选性

---

5. 结构化输出设计

5.1 为什么使用结构化输出？

1. 数据一致性：确保输出格式统一，便于后续处理
2. 信息丰富：不仅包含文本回答，还包含评分、推荐等结构化信息
3. 易于解析：使用Pydantic自动验证和解析
4. 可扩展性：可以轻松添加新字段

5.2 输出格式示例

用户问题：请解说《肖申克的救赎》

结构化输出：

{
    "answer": "《肖申克的救赎》是一部1994年上映的经典剧情片，由弗兰克·德拉邦特执导，蒂姆·罗宾斯和摩根·弗里曼主演。影片讲述了银行家安迪因被误判谋杀而入狱，在肖申克监狱中遇到了瑞德。安迪利用自己的金融知识帮助监狱长处理财务，同时策划越狱。经过19年的努力，安迪成功越狱并获得自由。\n\n这部电影深刻探讨了希望、自由、友谊和救赎的主题。影片通过安迪和瑞德的友谊，展现了人性在困境中的坚韧和希望的力量。导演运用精湛的叙事技巧和镜头语言，将监狱生活描绘得真实而深刻。\n\n《肖申克的救赎》被誉为电影史上的经典之作，豆瓣评分9.7分，是值得反复观看的佳作。",
    "movie_name": "肖申克的救赎",
    "category": "电影介绍",
    "key_points": [
        "1994年上映的经典剧情片",
        "导演：弗兰克·德拉邦特",
        "主演：蒂姆·罗宾斯、摩根·弗里曼",
        "主题：希望、自由、友谊、救赎",
        "豆瓣评分9.7分"
    ],
    "rating": 9.7,
    "recommendation": true,
    "related_movies": ["阿甘正传", "当幸福来敲门", "美丽人生"]
}

格式化后的用户友好输出：

《肖申克的救赎》是一部1994年上映的经典剧情片，由弗兰克·德拉邦特执导，蒂姆·罗宾斯和摩根·弗里曼主演。影片讲述了银行家安迪因被误判谋杀而入狱，在肖申克监狱中遇到了瑞德。安迪利用自己的金融知识帮助监狱长处理财务，同时策划越狱。经过19年的努力，安迪成功越狱并获得自由。

这部电影深刻探讨了希望、自由、友谊和救赎的主题。影片通过安迪和瑞德的友谊，展现了人性在困境中的坚韧和希望的力量。导演运用精湛的叙事技巧和镜头语言，将监狱生活描绘得真实而深刻。

《肖申克的救赎》被誉为电影史上的经典之作，豆瓣评分9.7分，是值得反复观看的佳作。

【关键要点】
1. 1994年上映的经典剧情片
2. 导演：弗兰克·德拉邦特
3. 主演：蒂姆·罗宾斯、摩根·弗里曼
4. 主题：希望、自由、友谊、救赎
5. 豆瓣评分9.7分

【评分】9.7/10
【推荐】强烈推荐

【相关电影推荐】
• 阿甘正传
• 当幸福来敲门
• 美丽人生

---

6. 完整实现代码

6.1 Agent核心类

from langchain_openai import ChatOpenAI
from langchain.memory import ConversationBufferMemory
from langchain_core.prompts import (
    MessagesPlaceholder,
    ChatPromptTemplate,
    SystemMessagePromptTemplate,
    HumanMessagePromptTemplate
)
from langchain_core.messages import HumanMessage, AIMessage
from typing import Optional

class MovieCommentatorAgent:
    """
    电影解说AI Agent
    
    功能：
    1. 回答电影相关问题
    2. 生成专业电影解说
    3. 支持多轮对话
    4. 支持结构化输出
    """

    def __init__(
        self, 
        api_key: Optional[str] = None, 
        model: str = "gpt-3.5-turbo", 
        use_deepseek: bool = False, 
        use_structured_output: bool = False
    ):
        """
        初始化电影解说Agent
        
        Args:
            api_key: API密钥（OpenAI或DeepSeek）
            model: 使用的模型名称
                - OpenAI: "gpt-3.5-turbo", "gpt-4" 等
                - DeepSeek: "deepseek-chat"
            use_deepseek: 是否使用DeepSeek API，默认False（使用OpenAI）
            use_structured_output: 是否使用结构化输出格式，默认False
        """
        if not api_key:
            raise ValueError("请提供API密钥")
        
        # 初始化输出解析器
        self.output_parser = MovieOutputParser()
        self.use_structured_output = use_structured_output

        # 初始化LLM（支持OpenAI和DeepSeek）
        if use_deepseek:
            # 使用DeepSeek API
            deepseek_base_url = "https://api.deepseek.com"
            self.llm = ChatOpenAI(
                model="deepseek-chat",
                openai_api_key=api_key,
                openai_api_base=deepseek_base_url,
                temperature=0.7,  # 创造性温度，0.7平衡创造性和准确性
                max_tokens=2000     # 最大输出token数
            )
        else:
            # 使用OpenAI API
            self.llm = ChatOpenAI(
                model=model,
                openai_api_key=api_key,
                temperature=0.7,
                max_tokens=2000
            )

        # 初始化记忆系统
        # ConversationBufferMemory保存完整的对话历史
        self.memory = ConversationBufferMemory(
            memory_key="chat_history",
            return_messages=True  # 返回消息对象而不是字符串
        )

        # 创建Agent（简化版，不使用工具）
        self.agent = self._create_agent()

    def _create_agent(self):
        """创建Agent（简化版，不使用工具）"""
        # 构建系统提示
        system_prompt = MOVIE_COMMENTATOR_SYSTEM_PROMPT
        
        # 根据模式添加格式说明
        if self.use_structured_output:
            system_prompt += "\n\n" + self.output_parser.get_format_instructions()
        else:
            system_prompt += "\n\n" + self.output_parser.get_simple_format_instructions()
        
        # 创建Prompt模板
        # 包含：系统消息、对话历史、用户输入
        prompt = ChatPromptTemplate.from_messages([
            SystemMessagePromptTemplate.from_template(system_prompt),
            MessagesPlaceholder(variable_name="chat_history"),  # 对话历史占位符
            HumanMessagePromptTemplate.from_template("{input}"),  # 用户输入
        ])

        # 创建简单的Agent类（不使用LangChain的AgentExecutor）
        # 这样可以更灵活地控制LLM调用和记忆管理
        class SimpleAgent:
            def __init__(self, llm, prompt, memory):
                self.llm = llm
                self.prompt = prompt
                self.memory = memory
            
            def invoke(self, inputs):
                """
                调用Agent处理用户输入
                
                Args:
                    inputs: 包含"input"键的字典
                
                Returns:
                    包含"output"键的字典
                """
                # 获取历史消息
                chat_history = self.memory.chat_memory.messages
                
                # 构建消息列表（系统消息 + 历史消息 + 用户消息）
                messages = self.prompt.format_messages(
                    chat_history=chat_history,
                    input=inputs["input"]
                )
                
                # 调用 LLM
                response = self.llm.invoke(messages)
                
                # 保存到记忆（用户消息和AI回复）
                self.memory.chat_memory.add_user_message(
                    HumanMessage(content=inputs["input"])
                )
                self.memory.chat_memory.add_ai_message(
                    AIMessage(content=response.content)
                )
                
                return {"output": response.content}
        
        return SimpleAgent(self.llm, prompt, self.memory)

    def ask(self, question: str, verbose: bool = True) -> str:
        """
        回答关于电影的问题
        
        Args:
            question: 用户问题
            verbose: 是否显示详细信息（关键要点、评分、推荐等），默认True
        
        Returns:
            回答内容（格式化后的文本）
        """
        try:
            # 调用Agent处理问题
            response = self.agent.invoke({"input": question})
            raw_output = response.get("output", str(response))
            
            # 如果使用结构化输出，解析并格式化
            if self.use_structured_output:
                try:
                    # 解析JSON响应
                    parsed_response = self.output_parser.parse(raw_output)
                    # 格式化为用户友好的文本
                    formatted = self.output_parser.format_response(
                        parsed_response, 
                        verbose=verbose
                    )
                    # 如果解析后的回答为空，返回原始输出
                    if not formatted or not formatted.strip():
                        return raw_output
                    return formatted
                except Exception as e:
                    # 如果解析失败，返回原始输出（静默处理）
                    return raw_output
            else:
                # 简单模式，直接返回原始输出
                return raw_output
        except Exception as e:
            return f"回答问题时出错：{str(e)}"

    def reset_memory(self):
        """重置对话记忆"""
        self.memory.clear()
        print("对话记忆已重置")

关键设计点：

1. SimpleAgent类：不使用LangChain的AgentExecutor，更灵活地控制流程
2. 记忆管理：每次调用后自动保存用户消息和AI回复
3. 双模式支持：根据use_structured_output参数选择输出模式
4. 错误处理：解析失败时返回原始输出，保证系统稳定性

6.2 交互模式

def interactive_mode(api_key: Optional[str] = None, use_deepseek: bool = False):
    """
    交互模式：提供命令行交互界面
    
    Args:
        api_key: API密钥（OpenAI或DeepSeek）
        use_deepseek: 是否使用DeepSeek API
    """
    print("=" * 70)
    print("🎬 电影解说AI Agent - 交互模式")
    print("=" * 70)
    print("\n提示：")
    print("  - 输入电影名称可以生成解说（如：请解说《肖申克的救赎》）")
    print("  - 输入问题可以询问电影信息（如：这部电影的导演是谁？）")
    print("  - 输入 'reset' 可以重置对话记忆")
    print("  - 输入 'quit' 或 'exit' 退出")
    print("=" * 70)

    if not api_key:
        print("\n❌ 错误：未提供API密钥")
        return

    try:
        # 创建Agent实例
        agent = MovieCommentatorAgent(
            api_key=api_key, 
            use_deepseek=use_deepseek,
            use_structured_output=True  # 使用结构化输出
        )
        print("\n✅ Agent已就绪，开始对话吧！\n")

        while True:
            # 获取用户输入
            user_input = input("\n你: ").strip()

            if not user_input:
                continue

            # 退出命令
            if user_input.lower() in ['quit', 'exit', '退出']:
                print("\n👋 再见！")
                break

            # 重置记忆命令
            if user_input.lower() == 'reset':
                agent.reset_memory()
                continue

            try:
                # 调用Agent回答问题
                response = agent.ask(user_input)
                print(f"\nAgent: {response}")
            except Exception as e:
                print(f"\n❌ 错误：{str(e)}")

    except KeyboardInterrupt:
        print("\n\n👋 再见！")
    except Exception as e:
        print(f"\n❌ 发生错误：{str(e)}")
        import traceback
        traceback.print_exc()

---

7. 使用示例

7.1 基础使用

from movies_agent import MovieCommentatorAgent

# 初始化Agent（使用DeepSeek API，推荐）
agent = MovieCommentatorAgent(
    api_key="your-deepseek-api-key",
    use_deepseek=True,
    use_structured_output=True  # 使用结构化输出
)

# 生成电影解说
commentary = agent.ask("请解说《肖申克的救赎》")
print(commentary)

# 回答问题（Agent会记住之前的对话）
answer = agent.ask("这部电影的导演是谁？")
print(answer)

# 继续对话
answer2 = agent.ask("这部电影的主题是什么？")
print(answer2)

# 重置记忆
agent.reset_memory()

7.2 完整对话示例

对话1：电影解说

你: 请解说《肖申克的救赎》

Agent: 《肖申克的救赎》是一部1994年上映的经典剧情片，由弗兰克·德拉邦特执导...

【关键要点】
1. 1994年上映的经典剧情片
2. 导演：弗兰克·德拉邦特
3. 主题：希望、自由、友谊、救赎
4. 豆瓣评分9.7分

【评分】9.7/10
【推荐】强烈推荐

对话2：基于历史的问答

你: 这部电影的导演是谁？

Agent: 这部电影的导演是弗兰克·德拉邦特（Frank Darabont）...

【关键要点】
1. 导演：弗兰克·德拉邦特
2. 1994年执导《肖申克的救赎》
3. 改编自斯蒂芬·金的小说

对话3：主题探讨

你: 这部电影的主题是什么？

Agent: 《肖申克的救赎》的主题主要包括希望、自由、友谊和救赎...

【关键要点】
1. 希望：安迪始终相信希望的力量
2. 自由：对自由的渴望和追求
3. 友谊：安迪和瑞德的深厚友谊
4. 救赎：通过坚持和努力实现自我救赎

7.3 命令行使用

# 使用DeepSeek API（推荐）
python movies_agent.py --use-deepseek

# 使用OpenAI API
python movies_agent.py

# 指定API密钥
python movies_agent.py --api-key your-api-key --use-deepseek

---

8. 项目优化建议

8.1 功能扩展

1. 集成真实API：使用TMDB、IMDb等电影API获取实时信息

   def search_movie_info(movie_name: str) -> dict:
       # 调用TMDB API
       response = requests.get(
           f"https://api.themoviedb.org/3/search/movie",
           params={"api_key": TMDB_API_KEY, "query": movie_name}
       )
       return response.json()
   

2. 多语言支持：支持英文、日文等电影解说

   def ask(self, question: str, language: str = "zh") -> str:
       # 根据language参数调整Prompt
       pass
   

3. 流式输出：使用流式输出提升用户体验

   def ask_stream(self, question: str):
       # 使用stream=True参数
       for chunk in self.llm.stream(messages):
           yield chunk.content
   

4. 批量处理：支持批量生成多个电影解说

   def batch_comment(self, movie_names: List[str]) -> List[str]:
       results = []
       for movie_name in movie_names:
           result = self.ask(f"请解说《{movie_name}》")
           results.append(result)
       return results
   

8.2 性能优化

1. 缓存机制：缓存常见电影信息减少API调用

   from functools import lru_cache
   
   @lru_cache(maxsize=100)
   def get_movie_info(movie_name: str) -> dict:
       # 缓存电影信息
       pass
   

2. 异步处理：支持异步请求提升响应速度

   import asyncio
   
   async def ask_async(self, question: str) -> str:
       # 异步调用LLM
       response = await self.llm.ainvoke(messages)
       return response.content
   

3. Token优化：优化Prompt长度减少Token消耗

   • 使用摘要Memory替代完整Memory
   • 压缩历史对话内容
   • 使用更简洁的Prompt

8.3 用户体验优化

1. 错误处理：完善的错误处理和用户提示

   try:
       response = agent.ask(question)
   except APIError as e:
       return f"API调用失败：{str(e)}，请稍后重试"
   except ParseError as e:
       return f"解析失败：{str(e)}，返回原始回答"
   

2. 进度提示：长时间处理时显示进度

   print("正在生成电影解说，请稍候...")
   response = agent.ask(question)
   print("生成完成！")
   

3. 个性化设置：允许用户自定义输出格式

   agent = MovieCommentatorAgent(
       api_key=api_key,
       verbose_format=True,  # 详细格式
       include_rating=True,  # 包含评分
       include_recommendation=True  # 包含推荐
   )
   

8.4 代码质量优化

1. 类型提示：完善类型提示提高代码可读性

   from typing import List, Optional, Dict, Union
   
   def ask(self, question: str, verbose: bool = True) -> str:
       # 明确的类型提示
       pass
   

2. 单元测试：添加单元测试保证代码质量

   def test_movie_parser():
       parser = MovieOutputParser()
       test_json = '{"answer": "test", "category": "电影介绍"}'
       result = parser.parse(test_json)
       assert result.answer == "test"
   

3. 日志记录：添加日志记录便于调试

   import logging
   
   logger = logging.getLogger(__name__)
   
   def ask(self, question: str) -> str:
       logger.info(f"用户问题：{question}")
       response = self.agent.invoke({"input": question})
       logger.info(f"Agent回复：{response}")
       return response
   

---

9. 总结与最佳实践

9.1 技术栈总结

本项目使用的核心技术栈：

电影解说AI Agent技术栈

┌──────────────┐
│  LangChain   │ → 应用框架（LLM调用、记忆管理）
└──────┬───────┘
       │
┌──────▼───────┐
│  ChatOpenAI  │ → LLM接口（OpenAI/DeepSeek）
└──────┬───────┘
       │
┌──────▼───────┐
│   Pydantic   │ → 数据验证和结构化输出
└──────┬───────┘
       │
┌──────▼───────┐
│   Memory     │ → 对话记忆管理
└──────┬───────┘
       │
┌──────▼───────┐
│   Parser     │ → JSON解析和格式化
└──────────────┘

9.2 最佳实践清单

✅ 结构化输出设计

1. 使用Pydantic模型：定义清晰的数据结构
2. 字段验证：使用Field添加验证规则和描述
3. 容错处理：解析失败时提供默认值
4. 格式化输出：将结构化数据转换为用户友好的文本

✅ Prompt设计

1. 角色定位明确：明确Agent的角色和能力
2. 格式说明清晰：提供具体的输出格式示例
3. 重要提示：列出关键注意事项
4. 迭代优化：根据实际效果不断改进Prompt

✅ 记忆管理

1. 选择合适的Memory类型：根据场景选择BufferMemory或SummaryMemory
2. 及时保存：每次对话后保存到记忆
3. 记忆重置：提供重置功能避免上下文过长

✅ 错误处理

1. 多策略解析：支持多种JSON格式提高成功率
2. 静默失败：解析失败时返回原始输出，不中断流程
3. 用户提示：API错误时给出友好的错误提示

✅ API选择

1. 成本考虑：DeepSeek API性价比高，适合开发测试
2. 性能考虑：GPT-4性能更好，适合生产环境
3. 灵活切换：代码支持两种API无缝切换

9.3 常见问题与解决方案

问题1: JSON解析失败

原因：LLM输出格式不符合预期

解决方案：

• 使用多策略解析（代码块、纯JSON、部分匹配）
• 提供清晰的格式说明和示例
• 解析失败时返回原始输出

问题2: 记忆过长导致Token消耗大

原因：ConversationBufferMemory保存完整历史

解决方案：

• 使用ConversationSummaryMemory替代
• 定期重置记忆
• 限制记忆长度

问题3: API调用失败

原因：网络问题、API密钥错误、速率限制

解决方案：

• 添加重试机制
• 检查API密钥有效性
• 实现速率限制和退避策略

问题4: 输出质量不稳定

原因：Temperature设置不当、Prompt不够清晰

解决方案：

• 调整Temperature（0.7平衡创造性和准确性）
• 优化Prompt，提供更多示例
• 使用更好的模型（GPT-4）

9.4 项目亮点

1. 结构化输出：使用Pydantic定义输出格式，信息丰富且易于解析
2. 双模式支持：支持结构化输出和简单文本输出
3. 多API支持：同时支持OpenAI和DeepSeek API
4. 智能解析：强大的JSON解析器，支持多种格式
5. 对话记忆：支持多轮对话，记住历史上下文
6. 用户友好：格式化输出，提供关键要点、评分、推荐等信息

9.5 未来发展方向

1. 集成真实API：使用TMDB、IMDb等电影API获取实时信息
2. 多模态支持：结合图像、视频等多模态输入
3. 个性化推荐：基于用户喜好推荐电影
4. 流式输出：使用流式输出提升用户体验
5. 批量处理：支持批量生成多个电影解说
6. Web界面：开发Web界面提供更好的用户体验

---

附录

A. 依赖安装

创建requirements.txt文件：

# LangChain核心库
langchain>=0.3.0
langchain-openai>=1.0.0
langchain-core>=0.3.0
langchain-community>=0.3.0

# OpenAI API（必须）
openai>=1.0.0

# 数据验证
pydantic>=2.0.0

# 其他工具
python-dotenv>=1.0.0
requests>=2.31.0

安装依赖：

pip install -r requirements.txt

B. 配置文件示例

创建.env文件：

# DeepSeek API（推荐）
DEEPSEEK_API_KEY=your-deepseek-api-key

# OpenAI API（可选）
OPENAI_API_KEY=your-openai-api-key

# 默认使用DeepSeek
USE_DEEPSEEK=True

C. 完整代码结构

my_agent/
├── movies_agent.py          # 主程序文件
├── requirements.txt         # 依赖文件
├── .env                     # 配置文件（可选）
└── README.md               # 项目说明

D. 快速开始

1. 安装依赖

   pip install -r requirements.txt
   

2. 设置API密钥

   # 在movies_agent.py中设置
   DEEPSEEK_API_KEY = "your-api-key"
   

3. 运行程序

   python movies_agent.py --use-deepseek
   

4. 开始对话

   你: 请解说《肖申克的救赎》
   Agent: [返回电影解说]
   

---

movies_agent.py完整代码：

"""电影解说系统输出格式定义和解析器"""
from typing import List, Optional, Literal
from pydantic import BaseModel, Field
from langchain_core.output_parsers import PydanticOutputParser
import json
import re

# 导入新版本LangChain API (>=0.3.0)
try:
    from langchain_openai import ChatOpenAI
    # 尝试从不同位置导入 ConversationBufferMemory
    try:
        from langchain.memory import ConversationBufferMemory
    except ImportError:
        try:
            from langchain_core.memory import ConversationBufferMemory
        except ImportError:
            from langchain_classic.memory import ConversationBufferMemory
    from langchain_core.prompts import (
        MessagesPlaceholder,
        ChatPromptTemplate,
        SystemMessagePromptTemplate,
        HumanMessagePromptTemplate
    )
    from langchain_core.messages import HumanMessage, AIMessage
except ImportError as e:
    error_msg = f"""
{'=' * 70}
❌ LangChain导入失败！
{'=' * 70}

【解决方案】

请安装新版本LangChain依赖：
    pip install -r requirements.txt

或者手动安装：
    pip install langchain>=0.3.0 langchain-openai>=1.0.0 langchain-core langchain-community openai

【错误详情】
{str(e)}

【验证安装】
安装后运行以下命令验证：
    python -c "from langchain_openai import ChatOpenAI; print('✓ 安装成功')"
{'=' * 70}
"""
    print(error_msg)
    raise ImportError("LangChain导入失败，请运行: pip install -r requirements.txt")

# 系统Prompt
MOVIE_COMMENTATOR_SYSTEM_PROMPT = """
你是一位专业的电影解说员，具有以下特点：

1. **专业素养**
   - 对电影艺术有深入理解
   - 熟悉各种电影类型和风格
   - 能够分析电影的深层含义

2. **解说风格**
   - 语言生动有趣，引人入胜
   - 结构清晰，逻辑严密
   - 既有专业分析，又通俗易懂

3. **内容要求**
   - 介绍电影基本信息（导演、演员、类型等）
   - 概述剧情（避免剧透关键情节）
   - 分析电影主题和艺术特色
   - 评价电影的艺术价值和观赏价值

4. **交互方式**
   - 回答用户关于电影的各种问题
   - 基于对话历史提供连贯的回答
   - 运用你的电影知识直接回答用户的问题

请始终以专业、友好、生动的语调与用户交流。
"""


class MovieResponse(BaseModel):
    """电影解说响应格式"""
    
    answer: str = Field(
        description="对用户问题的详细回答，这是主要回答内容，应该完整、准确、友好、生动"
    )
    
    movie_name: Optional[str] = Field(
        default=None,
        description="如果问题涉及特定电影，提供电影名称。如果不涉及特定电影，则为空"
    )
    
    category: Literal[
        "电影介绍",
        "剧情分析",
        "角色分析",
        "主题探讨",
        "技术评价",
        "推荐建议",
        "比较分析",
        "其他"
    ] = Field(
        default="其他",
        description="回答所属的分类"
    )
    
    key_points: Optional[List[str]] = Field(
        default=None,
        description="回答中的关键要点列表，用于快速了解核心内容"
    )
    
    rating: Optional[float] = Field(
        default=None,
        ge=0.0,
        le=10.0,
        description="如果涉及电影评分，提供评分（0-10分）。如果不涉及评分，则为空"
    )
    
    recommendation: Optional[bool] = Field(
        default=None,
        description="是否推荐观看这部电影。如果问题不涉及推荐，则为空"
    )
    
    related_movies: Optional[List[str]] = Field(
        default=None,
        description="相关的其他电影推荐列表，用户可能感兴趣的电影"
    )


class MovieOutputParser:
    """电影解说系统输出解析器"""
    
    def __init__(self):
        """初始化解析器"""
        self.parser = PydanticOutputParser(pydantic_object=MovieResponse)
        self.format_instructions = self.parser.get_format_instructions()
    
    def parse(self, text: str) -> MovieResponse:
        """解析LLM输出"""
        if not text or not text.strip():
            return MovieResponse(
                answer="",
                movie_name=None,
                category="其他",
                key_points=None,
                rating=None,
                recommendation=None,
                related_movies=None
            )
        
        # 首先尝试提取JSON（避免解析包含Schema说明的文本）
        json_str = self._extract_json(text)
        if json_str:
            try:
                data = json.loads(json_str)
                # 验证是否是有效的响应数据（不是Schema）
                if isinstance(data, dict):
                    # 如果包含answer字段，尝试创建响应
                    if "answer" in data:
                        try:
                            return MovieResponse(**data)
                        except Exception as e:
                            # 如果字段验证失败，使用answer字段作为回答
                            answer = data.get("answer", text)
                            return MovieResponse(
                                answer=answer,
                                movie_name=data.get("movie_name"),
                                category=data.get("category", "其他"),
                                key_points=data.get("key_points"),
                                rating=data.get("rating"),
                                recommendation=data.get("recommendation"),
                                related_movies=data.get("related_movies")
                            )
            except (json.JSONDecodeError, ValueError, TypeError) as e:
                pass
        
        # 尝试使用Pydantic parser解析（作为备选方案）
        try:
            parsed = self.parser.parse(text)
            return parsed
        except Exception:
            pass
        
        # 如果都失败了，创建一个默认响应，使用原始文本作为回答
        # 不打印错误信息，避免干扰用户体验
        return MovieResponse(
            answer=text,
            movie_name=None,
            category="其他",
            key_points=None,
            rating=None,
            recommendation=None,
            related_movies=None
        )
    
    def _extract_json(self, text: str) -> Optional[str]:
        """从文本中提取JSON内容"""
        # 清理文本，移除可能的格式说明
        text = text.strip()
        
        # 尝试提取代码块中的JSON
        json_patterns = [
            r"```json\s*(\{.*?\})\s*```",
            r"```\s*(\{.*?\})\s*```",
        ]
        
        for pattern in json_patterns:
            match = re.search(pattern, text, re.DOTALL)
            if match:
                json_str = match.group(1).strip()
                if self._is_valid_json_data(json_str):
                    return json_str
        
        # 尝试直接提取JSON对象（从第一个 { 到最后一个 }）
        if text.startswith("{"):
            # 找到最后一个 }
            last_brace = text.rfind("}")
            if last_brace > 0:
                json_str = text[:last_brace + 1]
                if self._is_valid_json_data(json_str):
                    return json_str
        
        # 尝试查找包含 answer 字段的 JSON
        json_match = re.search(r'\{[^{}]*"answer"[^{}]*\}', text, re.DOTALL)
        if json_match:
            # 尝试扩展匹配到完整的JSON对象
            start = text.rfind("{", 0, json_match.start() + 1)
            if start >= 0:
                # 找到匹配的结束括号
                brace_count = 0
                for i in range(start, len(text)):
                    if text[i] == "{":
                        brace_count += 1
                    elif text[i] == "}":
                        brace_count -= 1
                        if brace_count == 0:
                            json_str = text[start:i+1]
                            if self._is_valid_json_data(json_str):
                                return json_str
                            break
        
        return None
    
    def _is_valid_json_data(self, json_str: str) -> bool:
        """验证是否是有效的JSON数据（不是Schema定义）"""
        try:
            data = json.loads(json_str)
            if not isinstance(data, dict):
                return False
            # 如果是Schema定义（包含properties但不包含answer），返回False
            if "properties" in data and "required" in data and "answer" not in data:
                return False
            # 如果包含answer字段，是有效数据
            if "answer" in data:
                return True
            # 其他情况也尝试（可能是部分数据）
            return True
        except:
            return False
    
    def format_response(self, response: MovieResponse, verbose: bool = True) -> str:
        """格式化响应为友好的文本格式"""
        output_parts = []
        
        # 主要回答
        output_parts.append(response.answer)
        
        # 详细信息（仅在verbose模式下显示）
        if verbose:
            # 关键要点
            if response.key_points:
                output_parts.append("\n\n【关键要点】")
                for i, point in enumerate(response.key_points, 1):
                    output_parts.append(f"{i}. {point}")
            
            # 评分
            if response.rating is not None:
                output_parts.append(f"\n【评分】{response.rating}/10")
            
            # 推荐
            if response.recommendation is not None:
                rec_text = "强烈推荐" if response.recommendation else "不推荐"
                output_parts.append(f"【推荐】{rec_text}")
            
            # 相关电影
            if response.related_movies:
                output_parts.append("\n【相关电影推荐】")
                for movie in response.related_movies:
                    output_parts.append(f"• {movie}")
        
        return "\n".join(output_parts)
    
    def get_format_instructions(self) -> str:
        """获取格式说明，用于添加到prompt中"""
        return f"""请严格按照以下JSON格式输出你的回答（只输出JSON数据，不要输出Schema定义）：

{self.format_instructions}

输出示例（这是你需要输出的格式）：
{{
    "answer": "《肖申克的救赎》是一部1994年上映的经典剧情片，由弗兰克·德拉邦特执导...",
    "movie_name": "肖申克的救赎",
    "category": "电影介绍",
    "key_points": [
        "1994年上映的经典剧情片",
        "导演：弗兰克·德拉邦特",
        "主题：希望、自由、友谊、救赎",
        "豆瓣评分9.7分"
    ],
    "rating": 9.7,
    "recommendation": true,
    "related_movies": ["阿甘正传", "当幸福来敲门", "美丽人生"]
}}

重要提示：
1. **只输出JSON数据对象，不要输出Schema定义或格式说明**
2. answer字段是必需的，必须包含完整的回答内容
3. category必须是预定义的分类之一："电影介绍"、"剧情分析"、"角色分析"、"主题探讨"、"技术评价"、"推荐建议"、"比较分析"、"其他"
4. 如果问题不涉及特定电影，movie_name可以为null
5. key_points、rating、recommendation、related_movies可以为空或null
6. 用中文回答，语言要专业、友好、生动、引人入胜
7. 评分范围是0-10分，保留一位小数
8. **直接输出JSON，不要包含任何其他文字说明**"""
    
    def get_simple_format_instructions(self) -> str:
        """获取简化版格式说明（用于不需要结构化输出的场景）"""
        return """请用专业、友好、生动的语言回答用户的问题。
回答应该：
1. 结构清晰，逻辑严密
2. 语言生动有趣，引人入胜
3. 既有专业分析，又通俗易懂
4. 如果涉及电影，提供基本信息、剧情概述、主题分析和评价"""



class MovieCommentatorAgent:
    """
    电影解说AI Agent

    功能：
    1. 搜索电影信息
    2. 回答电影相关问题
    3. 支持多轮对话
    """

    def __init__(self, api_key: Optional[str] = None, model: str = "gpt-3.5-turbo", use_deepseek: bool = False, use_structured_output: bool = False):
        """
        初始化电影解说Agent

        Args:
            api_key: API密钥（OpenAI或DeepSeek）
            model: 使用的模型名称
                - OpenAI: "gpt-3.5-turbo", "gpt-4" 等
                - DeepSeek: "deepseek-chat"
            use_deepseek: 是否使用DeepSeek API，默认False（使用OpenAI）
            use_structured_output: 是否使用结构化输出格式，默认True
        """
        if not api_key:
            raise ValueError("请提供API密钥")
        
        # 初始化输出解析器
        self.output_parser = MovieOutputParser()
        self.use_structured_output = use_structured_output

        # 初始化LLM
        if use_deepseek:
            # 使用DeepSeek API
            deepseek_base_url = "https://api.deepseek.com"
            self.llm = ChatOpenAI(
                model="deepseek-chat",
                openai_api_key=api_key,
                openai_api_base=deepseek_base_url,
                temperature=0.7,
                max_tokens=2000
            )
        else:
            # 使用OpenAI API
            self.llm = ChatOpenAI(
                model=model,
                openai_api_key=api_key,
                temperature=0.7,
                max_tokens=2000
            )

        # 初始化记忆系统
        self.memory = ConversationBufferMemory(
            memory_key="chat_history",
            return_messages=True
        )

        # 创建工具
        self.tools = self._create_tools()

        # 创建Agent
        self.agent = self._create_agent()

    def _create_tools(self):
        """创建工具集"""
        # 不使用工具，直接依赖 LLM 的知识
        return []

    def _create_agent(self):
        """创建Agent（简化版，不使用工具）"""
        # 构建系统提示
        system_prompt = MOVIE_COMMENTATOR_SYSTEM_PROMPT
        if self.use_structured_output:
            system_prompt += "\n\n" + self.output_parser.get_format_instructions()
        else:
            system_prompt += "\n\n" + self.output_parser.get_simple_format_instructions()
        
        # 创建Prompt模板
        prompt = ChatPromptTemplate.from_messages([
            SystemMessagePromptTemplate.from_template(system_prompt),
            MessagesPlaceholder(variable_name="chat_history"),
            HumanMessagePromptTemplate.from_template("{input}"),
        ])

        # 创建简单的链（不使用工具）
        class SimpleAgent:
            def __init__(self, llm, prompt, memory):
                self.llm = llm
                self.prompt = prompt
                self.memory = memory
            
            def invoke(self, inputs):
                # 获取历史消息
                chat_history = self.memory.chat_memory.messages
                
                # 构建消息列表
                messages = self.prompt.format_messages(
                    chat_history=chat_history,
                    input=inputs["input"]
                )
                
                # 调用 LLM
                response = self.llm.invoke(messages)
                
                # 保存到记忆
                self.memory.chat_memory.add_user_message(HumanMessage(content=inputs["input"]))
                self.memory.chat_memory.add_ai_message(AIMessage(content=response.content))
                
                return {"output": response.content}
        
        return SimpleAgent(self.llm, prompt, self.memory)

    def ask(self, question: str, verbose: bool = True) -> str:
        """
        回答关于电影的问题

        Args:
            question: 用户问题
            verbose: 是否显示详细信息（关键要点、评分、推荐等），默认True

        Returns:
            回答内容
        """
        try:
            response = self.agent.invoke({"input": question})
            raw_output = response.get("output", str(response))
            
            # 如果使用结构化输出，解析并格式化
            if self.use_structured_output:
                try:
                    parsed_response = self.output_parser.parse(raw_output)
                    formatted = self.output_parser.format_response(parsed_response, verbose=verbose)
                    # 如果解析后的回答为空，返回原始输出
                    if not formatted or not formatted.strip():
                        return raw_output
                    return formatted
                except Exception as e:
                    # 如果解析失败，返回原始输出（静默处理，不打印错误）
                    return raw_output
            else:
                return raw_output
        except Exception as e:
            return f"回答问题时出错：{str(e)}"

    def reset_memory(self):
        """重置对话记忆"""
        self.memory.clear()
        print("对话记忆已重置")


def interactive_mode(api_key: Optional[str] = None, use_deepseek: bool = False):
    """交互模式

    Args:
        api_key: API密钥（OpenAI或DeepSeek）
        use_deepseek: 是否使用DeepSeek API
    """
    print("=" * 70)
    print("🎬 电影解说AI Agent - 交互模式")
    print("=" * 70)
    print("\n提示：")
    print("  - 输入电影名称可以生成解说（如：请解说《肖申克的救赎》）")
    print("  - 输入问题可以询问电影信息（如：这部电影的导演是谁？）")
    print("  - 输入 'reset' 可以重置对话记忆")
    print("  - 输入 'quit' 或 'exit' 退出")
    print("=" * 70)

    if not api_key:
        print("\n❌ 错误：未提供API密钥")
        print("请在main函数中设置api_key")
        return

    try:
        agent = MovieCommentatorAgent(api_key=api_key, use_deepseek=use_deepseek)
        print("\n✅ Agent已就绪，开始对话吧！\n")

        while True:
            user_input = input("\n你: ").strip()

            if not user_input:
                continue

            if user_input.lower() in ['quit', 'exit', '退出']:
                print("\n👋 再见！")
                break

            if user_input.lower() == 'reset':
                agent.reset_memory()
                continue

            try:
                response = agent.ask(user_input)
                print(f"\nAgent: {response}")
            except Exception as e:
                print(f"\n❌ 错误：{str(e)}")

    except KeyboardInterrupt:
        print("\n\n👋 再见！")
    except Exception as e:
        print(f"\n❌ 发生错误：{str(e)}")
        import traceback
        traceback.print_exc()


if __name__ == "__main__":
    import argparse

    # 在main函数中直接设置API密钥
    # 支持OpenAI和DeepSeek
    DEEPSEEK_API_KEY = "sk-82bbcd7562414210891e7f50e0d1ef66"  # DeepSeek API密钥
    OPENAI_API_KEY = None  # OpenAI API密钥（如果需要使用OpenAI，请设置）

    # 选择使用的API（True=DeepSeek, False=OpenAI）
    USE_DEEPSEEK = True

    parser = argparse.ArgumentParser(description="电影解说AI Agent")
    parser.add_argument(
        "--api-key",
        type=str,
        help="API密钥（可选，会覆盖main函数中的设置）"
    )
    parser.add_argument(
        "--use-deepseek",
        action="store_true",
        help="使用DeepSeek API（默认使用main函数中的设置）"
    )

    args = parser.parse_args()

    # 确定使用的API密钥和类型
    if args.api_key:
        api_key = args.api_key
        use_deepseek = args.use_deepseek if args.use_deepseek else USE_DEEPSEEK
    else:
        if USE_DEEPSEEK:
            api_key = DEEPSEEK_API_KEY
            use_deepseek = True
        else:
            api_key = OPENAI_API_KEY
            use_deepseek = False

    if not api_key:
        print("❌ 错误：未设置API密钥")
        print("请在main函数中设置DEEPSEEK_API_KEY或OPENAI_API_KEY")
        exit(1)

    interactive_mode(api_key=api_key, use_deepseek=use_deepseek)