从一条新闻到一个自主信息处理Agent

昨天（2026年5月29日）凌晨，I-95南向靠近Quantico海军陆战队基地路段发生严重车祸，5人死亡、30多人受伤。当我在手机上看到这条推送时，脑子里想的不是事故本身——而是：如果我们能让Agent自动完成信息收集、交叉验证、报告生成，记者和应急人员可以节省多少时间？

这条新闻来自WTOP，是一家本地广播电台。但实际救援中，需要从警方推特、医院声明、交通摄像头、目击者社交媒体等多个来源拼出完整拼图。每个来源可能互相矛盾（比如有的说“30多人受伤”，有的说“35人”）。手动验证要花大量精力，而Agent可以并行完成这些任务。

架构拆解：三个Agent协作

为了处理这类突发事件，我设计了三个专用Agent，按顺序和并行两种模式工作。

1. 信息收集Agent — 主动抓取+被动订阅

• 工具：搜索引擎API（Tavily）、社交媒体API（Twitter/Facebook）、政府RSS订阅（Virginia State Police）
• 任务：以初始新闻中的关键实体（“I-95 Stafford County”、“2026-05-29”、“fatal crash”）作为种子，搜索相关报道、官方声明、目击视频
• 输出：一个结构化的数据集合，每条包含来源、时间戳、原始文本片段

2. 验证Agent — 交叉比对与置信度评分

• 规则：每个关键字段（死亡人数、受伤人数、事故时间、地点）从至少两个独立来源验证
• 冲突解决：如果来源不一致，按权威度加权（警方公告 > 主流媒体 > 社交媒体 > 个人推文）。同时标记置信度（高/中/低）。
• 输出：一个已验证的事实表，附带冲突记录

3. 摘要Agent — 生成结构化报告

• 输入：验证后的事实表 + 原始时间线
• 任务：按“概要→关键数据→信息来源→时间线→影响”结构生成报告
• 考虑输出格式：JSON（供内部系统消费）和Markdown（供人类阅读）

核心流程图

初始新闻触发
     ↓
信息收集Agent（并行）
 ├─ 搜索Tavily获取最新报道
 ├─ 爬取警方推特
 ├─ 检查本地新闻RSS
 └─ 提取地理坐标和时间
     ↓
验证Agent（串行+并行）
 ├─ 死亡人数: 来源A=5, 来源B=5, 来源C=4 → 加权平均 → 置信度高
 ├─ 受伤人数: 来源A=30+, 来源B=35, 来源C=32 → 取中位数32，置信度中
 └─ 原因: 施工区减速 → 所有来源一致 → 置信度高
     ↓
摘要Agent（LLM调用）
 └─ 生成最终报告（JSON+Markdown）

关键实现细节和踩坑记录

细节1：如何避免重复抓取

• 给每个内容片段生成哈希（基于URL+发布时间），存入临时存储。下一次搜索时先检查哈希集合，跳过已处理的内容。

细节2：时间窗口管理

• 事故发生在2:35 AM，搜索范围从2:30 AM开始到当前时间。之后每15分钟增量更新，直到事件热度下降。

踩坑1：API限流

• Tavily和Twitter API都有速率限制。实现时增加指数退避重试，并设置最大并发数（我设为3）。

踩坑2：信息冲突处理

• 试运行时发现受伤人数在多个来源中差异较大（30 vs 35 vs 32）。我的观点：在这种场景下不要盲目相信多数。应该优先采纳院方或警方直接发布的数字（如果有的话）。如果只有媒体，采用中位数或范围（如“30-35人”）。

踩坑3：LLM幻觉

• 摘要Agent调用GPT-4时，曾自动补全“事故原因是司机疲劳驾驶”之类原文没有的信息。需要强制LLM只输出已验证字段，未验证字段标注“待核实”。

简化版动手实现（基于LangGraph）

下面是一个极简版Agent工作流，用Python伪代码展示核心逻辑。

from langgraph.graph import StateGraph, END

class NewsState:
    initial_news: str
    raw_sources: list = []
    verified_facts: dict = {}
    final_report: str = ""

def collect_sources(state: NewsState) -> NewsState:
    # 调用Tavily搜索，获取top 5结果
    results = tavily.search(state.initial_news, max_results=5)
    state.raw_sources = results
    return state

def verify_sources(state: NewsState) -> NewsState:
    facts = extract_and_validate(state.raw_sources)  # 内部逻辑：去重、冲突解决
    state.verified_facts = facts
    return state

def generate_report(state: NewsState) -> NewsState:
    prompt = f"""根据已验证事实{state.verified_facts}，生成突发事件简报。只使用已验证字段，未验证的写'待核实'。"""
    report = call_llm(prompt)
    state.final_report = report
    return state

# 构建图
graph = StateGraph(NewsState)
graph.add_node("collect", collect_sources)
graph.add_node("verify", verify_sources)
graph.add_node("report", generate_report)
graph.set_entry_point("collect")
graph.add_edge("collect", "verify")
graph.add_edge("verify", "report")
graph.add_edge("report", END)
app = graph.compile()

# 执行
state = NewsState(initial_news="5 dead, dozens injured in crash on I-95 southbound in Stafford Co.")
result = app.invoke(state)
print(result.final_report)

这对开发者意味着什么

突发事件处理只是Agent在实时信息领域的一个应用。同样的模式可以套用到舆情监控、竞品动态、金融事件响应中。关键收获是：不要试图用一个Agent包揽所有事情，拆分成专门的Agent+明确的冲突解决策略，才能让系统可靠。 如果你的产品需要从多个来源自动聚合并验证信息，今天提到的三Agent架构是一个非常实用的起点。

另外注意，验证Agent是整个系统的守门人——没有它，信息收集Agent会收集垃圾，摘要Agent会输出垃圾。花时间在验证逻辑上，比花在优化摘要语法上重要得多。

（本文中涉及的真实事故新闻仅作技术案例，向死伤者致哀。）