问题背景：为什么需要票房分析Agent

上周WSJ报道《曼达洛人与格罗古》以1.02亿美元开画，但放在整个电影工业里，这个数据需要横向对比、历史趋势、院线排片等诸多维度才能变成可操作的判断。如果你在追踪多个电影项目或做二级市场分析，手动从不同来源（WSJ、Box Office Mojo、IMDb）收集数据再写报告，一天就废了。

Agent的核心价值不是替代记者，而是把多步、跨工具、需要记忆的任务自动化。下面用这个票房案例拆解一个完整Agent的四个组件。

Agent架构拆解

典型的任务规划Agent包含四层：规划层、工具层、记忆层、执行层。

1. 规划层（分解任务）

给Agent一个指令：“分析《曼达洛人与格罗古》的开画票房，对比历史星战作品，生成市场简报。”规划器会把它拆成：

  • 从WSJ抓取最新票房数据（注意可能被付费墙屏蔽）
  • 从Box Office Mojo获取历史星战开画对比
  • 调用LLM总结趋势并生成Markdown报告
  • 检查数据完整性，如果抓取失败则重试或换源

2. 工具层（每个步骤的接口）

工具就像Agent的手和眼。我通常用这几种工具组合：

  • fetch_webpage(url)：带User-Agent模拟和重试，返回文本
  • query_database(sql)：如果本地有历史票房库，可快速查询
  • call_llm(prompt, model)：分析、生成报告
  • write_file(content, path)：输出最终文件

这里第一个工具踩坑最多——直接请求WSJ返回状态码403或付费墙摘要。你需要在请求时添加?format=amp或尝试textise前缀。或者更现实：改用公开API（如OMDb）或新闻聚合器（如NewsAPI）。

3. 记忆层（状态跟踪）

Agent需要记住已经做了哪些步骤、取到了什么数据。我用全局字典memory记录：

  • memory['current_data']：当前抓取结果
  • memory['failed_tools']：失败的工具列表，用于下次跳过
  • memory['cache']：避免重复调用LLM

真实场景中，记忆还可以持久化到文件，以便Agent崩溃后恢复。

4. 执行层（循环与重试）

每次执行一个计划步骤，检查结果，如果失败则调整计划再试。关键代码模式：

python
1 2 3 4 5 6 7 8 9 10 11 
max_retries = 3
for attempt in range(max_retries):
    result = execute_step(step)
    if result.success:
        memory['steps_result'][step.id] = result.data
        break
    else:
        memory['failed_tools'].append(step.tool_name)
        if attempt == max_retries - 1:
            fallback_step = replan(step, memory)
            execute_step(fallback_step)

核心流程图（文字版）

  1. User Input → Planning（LLM分解）
  2. 依次执行：Fetch WSJ → 解析摘要 → 提取数字和评语
  3. Fetch Box Office Mojo（抓取历史星战电影数据表）
  4. 合并数据存入memory
  5. LLM分析：对比增长率、市场份额、同档期竞品
  6. 输出Markdown报告

整个过程可以串行或并行（用asyncio），但新手建议先串行，保证每一步记忆准确。

关键实现细节与踩坑记录

1. 付费墙是最大敌人

WSJ等专业媒体要求订阅。我试过三种方案：

  • textise dot iitty（文本模式，但经常被禁）
  • 使用NewsAPI的摘要字段（免费版只有标题，付费版略好）
  • 直接用Box Office Mojo（无墙，但更新慢两三天）
    推荐组合：先试新闻数据API（比如GNews），如果拿不到具体数字再回退到公开票房榜。把你的Agent设计成“多源熔断”——一个源失败自动切下一个。

2. LLM数字解析的幻觉

让GPT提取“$102 million”很容易，但对比历史数据时，它可能自己编一个“2005年星战前传开画1.2亿”。解决办法：

  • 让Agent优先把数字写成到Python字典里，由代码计算趋势（例如current / previous - 1），LLM只做描述性总结。
  • 记忆层里缓存来源URL，输出时附上引用。

3. 成本控制

一次分析可能要调用LLM 3-5次。建议用GPT-4o-mini做规划，GPT-4o做关键总结。工具层里可以加一个cost_tracker，超过预算自动切换到更便宜的模型。

简化版动手实现

以下代码可以跑在本地，用NewsAPI模拟WSJ数据，用Box Office Mojo的CSV作为历史数据。

python
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 
import requests
import json
import openai
from typing import Dict, List

class BoxOfficeAgent:
    def __init__(self, api_key: str):
        self.api_key = api_key
        self.memory = {
            'current_data': {},
            'historical_data': [],
            'failed_tools': set()
        }
        self.openai_client = openai.OpenAI(api_key=openai.api_key)
    
    def fetch_news_data(self, film_name: str) -> Dict:
        if 'fetch_news' in self.memory['failed_tools']:
            return self.memory.get('fallback_data', {})
        try:
            # 实际使用NewsAPI或其他源
            url = 'https://newsapi.org/v2/everything'
            params = {
                'q': film_name + ' box office',
                'apiKey': self.api_key,
                'pageSize': 1
            }
            resp = requests.get(url, params=params, timeout=10)
            resp.raise_for_status()
            # 简化：直接返回假设数据
            self.memory['current_data'] = {
                'opening_weekend': 102_000_000,  # 模拟WSJ数字
                'source': 'NewsAPI',
                'date': '2026-05-24'
            }
            return self.memory['current_data']
        except Exception as e:
            self.memory['failed_tools'].add('fetch_news')
            self.memory['fallback_data'] = {'opening_weekend': 102_000_000, 'note': 'fallback'}
            return self.memory['fallback_data']
    
    def fetch_historical_data(self) -> List[Dict]:
        # 模拟历史票房
        return [
            {'film': 'The Force Awakens', 'opening': 247_000_000},
            {'film': 'The Last Jedi', 'opening': 220_000_000},
            {'film': 'Rise of Skywalker', 'opening': 177_000_000}
        ]
    
    def analyze_with_llm(self, current: Dict, historical: List) -> str:
        prompt = f"""当前电影《曼达洛人与格罗古》开画票房{current['opening_weekend']:,}美元。
历史星战电影开画：{', '.join([f'{h["film"]}: {h["opening"]:,}' for h in historical])}。
请用简洁的市场分析报告格式回答：
1. 与历史前三对比的百分比
2. 可能的因素（系列剧效应、间隔年等）
3. 这句定性的总结
"""
        response = self.openai_client.chat.completions.create(
            model="gpt-4o-mini",
            messages=[{"role": "user", "content": prompt}]
        )
        return response.choices[0].message.content
    
    def run(self, film_name: str):
        current = self.fetch_news_data(film_name)
        historical = self.fetch_historical_data()
        report = self.analyze_with_llm(current, historical)
        # 写入文件
        with open('box_office_report.md', 'w') as f:
            f.write(f"# {film_name} 票房简报\n\n")
            f.write(f"开画：${current['opening_weekend']/1_000_000:.0f}M\n")
            f.write(report)
        print("报告已生成：box_office_report.md")

# 使用示例
agent = BoxOfficeAgent(api_key="your_newsapi_key_here")
agent.run("The Mandalorian and Grogu")

注：你需要一个NewsAPI key（免费可申请）和OpenAI API key。如果不想付费，可以用免费LLM（如本地运行的Llama 3）替换analyze_with_llm部分。

我的个人观点

这种票房Agent真正有价值的场景不是写新闻，而是对冲基金和制片厂做实时市场判断。它把人力从收集数据解放到策略层面。但现有的Agent框架（如LangChain、CrewAI）对工具失败的处理都太轻量——直接抛异常。你需要像上面那样手动实现“熔断+回退”才能在生产环境生存。另外，记忆层不要全依赖LLM上下文，丢到SQLite里更稳。

下次再遇到类似新闻，不妨试试自己写一个Agent，而不是只读报道。