ʚ 𝙈𝙞𝙣𝙩₊✰‧˚在智能体框架中，事实获取（如从文档提取的结构化信息）与前端的传递是通过事件驱动机制和状态同步实现的，核心依赖异步事件流、结构化数据格式和前后端协议约定ʚ 𝙈𝙞𝙣𝙩₊✰‧˚

以下是具体实现细节：

一、事实获取的源头：后端如何生成事实数据

1.结构化事实数据的生成

🕊️事实提取的结果通过 ExtractedOutput 模型约束格式：

class ExtractedOutput(BaseModel):
    items: List[ExtractedItem]  # 提取的事实列表
    has_content: bool  # 是否有有效事实

class ExtractedItem(BaseModel):
    query: str  # 对应的子查询
    doc_idx: int  # 文档索引
    content: str  # 提取的事实内容

🥝例如：用户查询 “北京的人口和面积”，提取的事实可能是 [{"query": "北京的人口", "doc_idx": 0, "content": "北京常住人口2184万"}, ...]，该结构确保事实数据的规范性，便于前端解析

2.事实提取的事件触发

🕊️当 fact_extractor 完成提取后，会通过 yield 产出包含事实数据的事件

# fact_extractor/agent.py 核心逻辑
async def _run_async_impl(self, ctx: InvocationContext) -> AsyncGenerator[Event, None]:
    # 提取事实（省略具体逻辑）
    extracted_result = ExtractedOutput(items=..., has_content=True)
    # 生成包含事实数据的事件
    yield Event(
        content=types.Content(parts=[types.Part(text=extracted_result.json())], role="system"),
        custom_metadata={"event": "fact_extracted"},  # 标记事件类型
        timestamp=time.time()
    )

🥝事件中 content 字段存储序列化的事实数据（JSON 字符串），custom_metadata 标记事件类型（如 “fact_extracted”），方便前端识别

3.事实数据的结构化生成

1.数据模型定义（基于 Pydantic）

🕊️通过 Pydantic 模型强制约束事实数据的格式

from pydantic import BaseModel
from typing import List

class ExtractedItem(BaseModel):
    query: str  # 对应子查询（如"北京人口多少"）
    doc_idx: int  # 来源文档索引（用于关联原始文档）
    content: str  # 提取的事实内容（如"北京常住人口2184万"）
    confidence: float  # 置信度（0-1，评估事实可靠性）

class ExtractedOutput(BaseModel):
    items: List[ExtractedItem]  # 事实列表
    has_content: bool  # 是否有有效事实（用于前端判断显示逻辑）

🥝作用：确保后端生成的事实数据字段完整、类型统一，避免因格式混乱导致前端解析失败

2.事实提取逻辑

🕊️当 fact_extractor 子代理处理文档时，会调用大模型（如 GPT-4）提取与查询相关的信息，并映射到上述模型：

# 伪代码：调用大模型提取事实
def extract_facts_from_doc(query: str, doc_content: str) -> ExtractedOutput:
    prompt = f"从文档中提取回答'{query}'的事实：{doc_content}"
    llm_response = llm_client.generate(prompt)  # 调用大模型API
    # 解析大模型输出，映射到ExtractedOutput
    items = [
        ExtractedItem(
            query=query,
            doc_idx=0,
            content=parsed_content,
            confidence=0.92
        )
        for parsed_content in parse_llm_output(llm_response)
    ]
    return ExtractedOutput(items=items, has_content=len(items) > 0)

3.事件封装与元数据标记

🕊️生成事实数据后，后端会将其封装为 Event 对象，添加元数据用于前端识别：

from datetime import datetime
from typing import AsyncGenerator

async def _run_async_impl(self, ctx: InvocationContext) -> AsyncGenerator[Event, None]:
    # 提取事实（省略具体逻辑）
    extracted_result = extract_facts_from_doc(...)
    
    # 1. 序列化事实数据为JSON
    fact_json = extracted_result.json()  # Pydantic模型自带的序列化方法
    
    # 2. 封装为Event对象
    fact_event = Event(
        invocation_id=ctx.invocation_id,  # 关联当前会话的调用ID
        author=self.name,  # 标记事件来源（如"fact_extractor_agent"）
        content=types.Content(
            parts=[types.Part(text=fact_json)],  # 存储序列化的事实数据
            role="system"  # 角色标记（区分用户/系统/代理消息）
        ),
        custom_metadata={
            "event_type": "fact_extracted",  # 事件类型（前端过滤用）
            "timestamp": datetime.now().timestamp()  # 时间戳（前端排序用）
        },
        partial=False  # 标记为完整事件（非流式片段）
    )
    
    # 3. 通过异步生成器产出事件
    yield fact_event
    
    # 4. 同步到会话状态（用于断连后恢复）
    await ctx.session_service.append_event(
        ctx.session,
        Event(
            actions=EventActions(
                state_delta={"latest_facts": extracted_result.items}  # 存入会话状态
            ),
            author="internal_state_manager"
        )
    )

• 关键技术点：
  • 序列化：使用 Pydantic 的 json() 方法，确保复杂对象（如列表、嵌套字典）能被正确转换为字符串。
  • 元数据：event_type 用于前端区分事件类型（如事实提取、回答生成、错误提示），避免无效处理。

二、后端到前端的传递路径：事件流的实时推送

🕊️事实数据通过异步事件流从后端推送到前端，依赖框架的会话服务和通信协议：

1.会话服务作为事件中转站

🥝所有事件（包括事实提取结果）会先通过 ctx.session_service.append_event 存入会话历史，确保数据可追溯：

# 存入会话，供前端拉取或实时推送
await ctx.session_service.append_event(
    ctx.session,
    Event(
        content=types.Content(parts=[types.Part(text=fact_json)], role="system"),
        custom_metadata={"event": "fact_extracted"}
    )
)

🫛会话服务（session_service）维护一个事件列表，记录整个对话过程中的所有事件（用户消息、代理输出、事实数据等）

2.实时推送：基于 SSE 或 WebSocket

🫛前端通过服务器发送事件（SSE） 或WebSocket 与后端建立长连接，实时接收事件流：

• SSE 实现：后端通过 HTTP 响应持续推送事件，前端通过 EventSource 监听：

  // 前端接收SSE
  const eventSource = new EventSource(`/stream/${sessionId}`);
  eventSource.addEventListener('fact_extracted', (e) => {
      const factData = JSON.parse(e.data);  // 解析事实数据
      updateUI(factData);  // 更新前端界面
  });

  from fastapi import FastAPI, Request
  from fastapi.responses import StreamingResponse
  import asyncio
  
  app = FastAPI()
  
  @app.get("/stream/{session_id}")
  async def stream_events(session_id: str, request: Request):
      """建立SSE连接，实时推送会话事件"""
      async def event_generator():
          # 从会话服务获取事件队列
          event_queue = session_service.get_event_queue(session_id)
          
          while True:
              # 检查客户端是否断开连接
              if await request.is_disconnected():
                  break
              
              # 从队列获取最新事件
              event = await event_queue.get()
              
              # 按SSE格式封装（event:类型\ndata:内容\n\n）
              yield f"event: {event.custom_metadata['event_type']}\n"
              yield f"data: {event.content.parts[0].text}\n\n"
              
              # 标记事件已处理
              event_queue.task_done()
      
      # 返回流式响应，媒体类型为text/event-stream
      return StreamingResponse(
          event_generator(),
          media_type="text/event-stream"
      )

  🥝核心：事件类型（fact_extracted）作为前端过滤条件，只处理与事实相关的推送

智能体系统通常采用 SSE 而非 WebSocket，原因是：

3.数据格式：JSON 序列化与反序列化

🕊️事实数据在传递前会序列化为 JSON 字符串（通过 extracted_result.json()），确保跨语言传输兼容性；前端接收后通过 JSON.parse 还原为 JavaScript 对象，直接用于渲染

4.会话管理与事件可靠性保障

🕊️为避免事实数据丢失（如网络波动），后端通过会话事件队列和状态持久化确保可靠性：

1.事件队列：每个会话（session_id）对应一个异步队列（asyncio.Queue），所有事件先进入队列，再通过 SSE 逐个推送，避免并发混乱。

2.状态持久化：关键事实数据会存入会话状态（ctx.session.state），前端可通过 HTTP 接口拉取历史数据：

🥝SSE 是单向通信（后端→前端），适合事实数据这类 “后端主动推送” 的场景，无需前端频繁发送请求。

🫛基于 HTTP 协议，兼容性更好（无需额外端口，可穿透大部分防火墙）

• 三、前端如何处理与展示事实数据

  1.数据解析与验证

  🥝前端根据后端定义的 ExtractedOutput 结构解析数据，确保字段匹配

• function handleFactExtracted(data) {
      const { items, has_content } = data;
      if (!has_content) {
          showMessage("未提取到有效事实");
          return;
      }
      // 验证每个事实的字段
      items.forEach(item => {
          if (!item.query || !item.content) {
              console.warn("无效事实格式:", item);
          }
      });
      // 存储到前端状态管理（如Redux/Vuex）
      store.dispatch({ type: 'ADD_FACTS', payload: items });
  }

  2.与用户交互的联动

• 前端可基于事实数据提供交互功能，例如：

• 允许用户点击某个事实，查看对应的原始文档（通过 doc_idx 关联）。
• 若事实不足（has_content: false），前端显示 “正在补充搜索...” 的加载状态。

四、关键技术保障：确保传递的可靠性与实时性

1.事件类型的标准化

🕊️后端通过 custom_metadata 定义事件类型（如 fact_extracted、answerDelta、reasoningLogEnd），前端根据类型区分处理逻辑，避免混淆

2.会话状态的一致性

🥝事实数据同时存入会话状态（ctx.session.state），即使前端断连后重连，也可通过拉取历史事件恢复数据

# 后端：将事实存入会话状态
await ctx.session_service.append_event(
    ctx.session,
    Event(
        actions=EventActions(state_delta={"facts": extracted_result.items}),
        author="fact_extractor_internal"
    )
)

3.异步非阻塞设计

后端通过 async/await 和异步生成器（AsyncGenerator）确保事实提取和事件推送不阻塞主线程；前端通过事件监听实现非阻塞接收，避免界面卡顿

五、总结：事实获取与前端传递的完整链路

1.结构化数据传输：

🥝后端通过 Pydantic 模型约束事实数据格式，前端通过解析验证确保兼容性，解决了 “数据格式混乱导致的前后端对接失败” 问题

2.事件驱动的实时性：

🫛基于 SSE 协议的单向流推送，实现事实数据 “产生即推送”，避免前端轮询带来的资源浪费，确保用户能实时看到提取进度

3.可靠性保障机制：

会话事件队列确保事件有序处理，状态持久化支持断连后数据恢复，解决了 “网络不稳定导致的数据丢失” 问题

4.分层职责设计：

后端：负责事实提取、事件封装、协议实现

前端：负责数据解析、状态管理、UI 展示。
分层设计降低了耦合度，便于单独扩展（如替换通信协议、修改 UI 样式）

5.结构化事实生成 → 事件封装与元标记 → SSE实时推送 → 前端解析验证 → 状态存储与UI渲染

六、致谢

谢谢大家的阅读，很多不足支出，欢迎大家在评论区指出，如果我的内容对你有帮助，

可以点赞 ， 收藏 ，大家的支持就是我坚持下去的动力！

请赐予我平静，去接受我无法改变的 ：赐予我勇气，去改变我能改变的。