实时协作是企业项目管理工具里最难做的功能之一，不是因为它的需求复杂，而是因为它的工程约束极其苛刻：多个用户同时操作同一份数据，每个操作都需要在毫秒级内传播给其他参与者，同时还要保证数据的一致性不被破坏。

在这个场景里引入 AI Agent，挑战又上了一层楼。Agent 既是数据的消费者（需要实时感知协作现场的状态），又可能是数据的生产者（需要在协作流中插入 AI 生成的内容）。如何在高频事件流中保持 Agent 的状态准确、响应及时，同时避免 Agent 的写入操作破坏协作的一致性，是这个方向的核心工程问题。

---

一、实时协作的事件流模型

理解这个场景的工程问题，需要先建立一个清晰的事件流模型。

在实时协作系统中，所有的操作都被建模为事件。用户 A 在画布上打了一个点，是一个事件；用户 B 回复了这个打点，是一个事件；PM 修改了某个任务的截止日期，也是一个事件。这些事件按时间顺序形成一个不可变的日志序列。

时间轴 →
事件1: user_A 在坐标(120,340)打点，附文字"这里颜色有问题"
事件2: user_B 回复事件1："已记录，下次迭代修复"
事件3: user_A 关闭打点对话
事件4: user_C 修改任务#231 截止日期 2025-06-20 → 2025-06-25
事件5: system 触发自动提醒：任务#231 的下游任务#235 需关注

这个事件流有几个关键性质：

• 有序性：事件有全局时间戳，顺序不可更改

• 不可变性：事件一旦写入就不能修改，只能追加新事件来"撤销"旧事件

• 广播性：每个事件需要推送给所有当前在线的协作参与者

AI Agent 在这个模型里的位置，是事件流的消费者和有限制的生产者。

---

二、Agent 作为事件流消费者

Agent 消费事件流的目的是维护一个关于当前协作状态的"工作记忆"，用来支撑上下文感知的响应。

这里有一个核心设计决策：Agent 不应该试图消费完整的历史事件流。一个活跃的协作项目可能在一天内产生数千个事件，全部塞给 LLM 是不现实的。合理的做法是滑动窗口 + 摘要折叠：

class AgentContextBuilder:
    WINDOW_SIZE = 50  # 近期事件的滑动窗口大小
    
    def build_context(self, event_stream: EventStream, 
                      current_time: datetime) -> AgentContext:
        
        # 近期事件直接保留（细粒度）
        recent_events = event_stream.get_recent(limit=self.WINDOW_SIZE)
        
        # 历史事件折叠为结构化摘要（粗粒度）
        historical_summary = self._summarize_history(
            event_stream.get_before(current_time, exclude_recent=True)
        )
        
        return AgentContext(
            historical_summary=historical_summary,
            recent_events=recent_events,
            current_participants=event_stream.get_active_participants(),
            open_discussions=event_stream.get_unresolved_threads(),
        )
    
    def _summarize_history(self, events: list) -> dict:
        # 按事件类型聚合，而不是逐条保留
        return {
            "total_changes": len(events),
            "key_decisions": self._extract_decisions(events),
            "resolved_issues": self._count_resolved(events),
            "task_status_snapshot": self._build_status_snapshot(events),
        }

这个设计的核心思路是：越近期的事件，越可能与当前的交互相关，保留细节；越久远的事件，细节价值递减，只保留结论。

---

三、多用户并发操作的冲突处理

实时协作最棘手的问题是并发冲突：用户 A 和用户 B 同时编辑同一个任务字段，最终状态应该是什么？

主流的解决方案是 OT（操作变换） 或 CRDT（无冲突复制数据类型）。两者在工程复杂度上差异很大，选择依据主要是数据结构的类型：

数据类型	适合方案	典型场景
富文本、协同文档	OT 或 CRDT（如 Yjs）	多人同时编辑文档内容
结构化字段（状态、日期）	Last-Write-Wins + 版本号	任务状态、截止日期修改
画布打点	CRDT（G-Set 或 OR-Set）	多人在同一画布添加/删除标注

AI Agent 在这个环节的特殊处理：Agent 的写入操作必须带有明确的来源标记，且优先级低于人类操作。

class AgentWriter:
    def write_event(self, event: Event) -> WriteResult:
        # Agent 写入的事件带有特殊来源标记
        event.source = EventSource.AI_AGENT
        event.priority = Priority.LOW  # 遇到人类操作冲突时，AI 操作自动回退
        
        # Agent 写入前必须检查是否有人类在同时操作同一对象
        if self.conflict_detector.has_concurrent_human_op(event.target_id):
            # 延迟写入，等待人类操作完成
            return self.defer_write(event, delay_ms=500)
        
        return self.event_store.append(event)

这个"人类操作优先"的原则，避免了 Agent 的自动内容与用户的手动编辑产生混乱。

---

四、状态同步的最终一致性保证

在分布式协作系统里，“实时同步"并不意味着"强一致性”。网络抖动、消息重排、客户端离线重连都可能导致不同参与者看到的状态短暂不一致。

对于 AI Agent 来说，这意味着它感知到的状态可能是一个"快照"，而不是完全实时的全局状态。工程上的应对策略：

1. 乐观读取，保守写入

Agent 读取状态时接受"最终一致"的数据（可能有秒级延迟），但写入时必须通过带版本号的条件写入来避免覆盖更新的状态：

def conditional_update(task_id: str, new_value: str, 
                       expected_version: int) -> bool:
    current = store.get(task_id)
    if current.version != expected_version:
        # 版本不匹配，说明有其他人在这期间修改了数据
        # Agent 放弃写入，重新读取最新状态后再决策
        return False
    store.update(task_id, new_value, version=expected_version + 1)
    return True

2. 状态快照与增量同步结合

当 Agent 实例重启或长时间离线后重新上线时，不应该试图从头重放所有历史事件（这个代价过于昂贵）。合理的做法是：

重新上线流程：
1. 加载最近一次持久化的状态快照（例如每小时生成一次）
2. 从快照时间点之后，重放增量事件，追赶到当前状态
3. 开始正常消费实时事件流

3. 幂等性设计

网络重传可能导致同一个事件被 Agent 处理多次。Agent 的所有处理逻辑必须是幂等的——处理同一个事件一次和处理十次，最终状态应该相同：

class IdempotentEventProcessor:
    def __init__(self):
        self.processed_events = set()  # 生产环境用 Redis 替代
    
    def process(self, event: Event) -> None:
        if event.id in self.processed_events:
            return  # 跳过重复事件
        
        self._handle(event)
        self.processed_events.add(event.id)

---

五、Agent 在协作现场的主动介入时机

Agent 在实时协作场景里，什么时候应该主动发言，什么时候应该保持沉默？这是一个需要仔细拿捏的用户体验问题。

几个比较适合 Agent 主动介入的时机：

（1）讨论出现明显的信息缺口时。例如，两个用户在讨论某个需求变更，但他们似乎都不知道这个变更与某个已存在的约束冲突——Agent 可以在这时补充相关上下文。

（2）讨论产生了一个明确的待办事项但没有被记录时。例如，用户 A 说"这个问题我们下周二确认一下"，但没有在任务系统里创建对应的任务——Agent 可以提示"检测到一个未记录的待办事项，是否需要我创建一条任务？"

（3）协作中产生了一个需要更广泛知会的决策时。例如，PM 在小组讨论里拍板了某个技术方案，但这个决策可能影响另一个团队——Agent 可以建议"此决策可能需要同步给后端团队，是否需要我生成一条通知？"

反过来，Agent 不应该介入的情况包括：用户之间正在进行创意讨论（不确定性高，Agent 的介入会打断思路）、用户明确表示不需要 AI 建议的时候、以及 Agent 对当前讨论的上下文理解程度不足的时候（模糊的理解比没有帮助更有害）。

---

六、性能与成本的平衡

实时事件流的消费意味着高频的 LLM 调用——如果每个事件都触发一次推理，成本和延迟都会迅速失控。

实践中有效的几个控制策略：

事件批处理：不对单个事件触发推理，而是积累一个小批次（例如 10 个事件或 5 秒，取先到者），批量处理后统一决策是否需要 Agent 介入。

分层过滤：用轻量规则（非 LLM）做第一层过滤，把明显不需要 AI 处理的事件提前排除。例如，"用户 A 打开了某个任务详情页"这类纯浏览行为，不需要流入 LLM 层。

响应缓存：对于相似的协作状态，Agent 的响应可以有限度地缓存。例如，"当某类阻塞出现时，Agent 的标准提示是什么"可以预生成模板，减少实时 LLM 调用。

在这些工程手段的配合下，实时协作场景的 AI Agent 运行成本可以控制在可接受范围内，同时保持响应的及时性。

国产智能体服务商 Bizfocus ADP 的画布打点协作功能，是此类实时协作场景落地的工程参考之一。其打点对话机制本质上就是一套事件流架构，结合 AI 助理，可以在协作现场实现上下文感知的智能介入。

---

七、小结

实时协作场景中的 AI Agent 工程，本质上是在两个已经很复杂的系统（实时协作系统 + AI Agent 系统）之间建立可靠的接口。事件流模型为 Agent 提供了感知协作现场的统一入口；冲突处理和状态同步机制保证了 Agent 的写入不会破坏协作的一致性；批处理和分层过滤策略则控制了引入 Agent 之后的成本开销。

把这三件事想清楚，Agent 在实时协作场景里才能从"锦上添花的功能"变成"团队实际依赖的协作角色"。