用 Pieces API 自动生成每日团队站会报告

设想一下这样的场景：周一上午 9 点。你端起咖啡，加入站会，然后突然意识到：你完全不知道自己上周做了什么。一个周末就把你的大脑缓存清空了。是重构 React 组件吗？还是那已经是两周前的事了？你隐约记得周四一直在和 TypeScript 错误较劲，但最终的解决方案到底是什么？你的 Git 提交信息写着 fix: update logic——谢谢过去的自己，真是太有帮助了。

问题在于：你的大脑并不是为了充当完美的活动日志而设计的。但你的电脑呢？它记得一切。这正是 PiecesOS 的用武之地。

创建一个应用，跟踪每天完成的工作。

在第 1 部分中，我们将完成：

连接 PiecesOS

获取工作流摘要

按天对摘要进行分组

实时保持所有内容更新

稍后我们再处理 UI。让我们一步一步来！

Dart SDK 3.8.1 或更高版本（flutter.dev）

Gemini API 密钥（将在第 2 部分中提到）

首先，让我们创建一个新的 Flutter 应用。

flutter create daily_recap_app

在开始使用 Pieces SDK 之前，需要先将其添加到 Dart 或 Flutter 项目中。打开项目的 pubspec.yaml 文件，在 dependencies: 下添加以下内容：

dependencies:
  # The Pieces SDK
  pieces_os_client:
    git:
      url: https://github.com/pieces-app/pieces-os-client-sdk-for-dart.git

web_socket_channel: ^3.0.1

该 SDK 是根据我们的 API 配置自动生成的，因此它主要由帮助应用与 Pieces API 通信的代码组成。

我们还添加了 web_socket_channel 依赖，用于连接 Pieces WebSocket。

第 1 步：连接 PiecesOS

好了，第一个挑战——我们究竟该如何与 PiecesOS 通信？

首先创建一个新文件 lib/services/pieces_os_service.dart。

没什么复杂的——只是设置我们的 API 客户端。

现在进入有趣的部分——真正建立连接：

  // existing code from above
  Future<Application> connectApplication() async {
    final seededApp = SeededTrackedApplication(
      name: ApplicationNameEnum.OPEN_SOURCE,  // Hey Pieces, I'm open source!
      platform: Platform.operatingSystem == "macos"
          ? PlatformEnum.MACOS
          : PlatformEnum.WINDOWS,
      version: "0.0.1",
    );

final connection = SeededConnectorConnection(
      application: seededApp,
    );

_context = await _connectorApi.connect(
      seededConnectorConnection: connection,
    );

print('Successfully connected to Pieces OS!');
    return _context!.application;
  }

简单来说，我们是在说：“嘿，Pieces，我是一个新应用，让我接入吧！”然后它会返回一个包含应用信息的上下文。

第 2 步：WebSocket

首次连接时，这个 WebSocket 会向我们发送 Pieces 中用户所有工作流摘要的 ID。之后，它只会发送新创建的摘要、数据发生更新的摘要或已删除摘要的 ID。

  // existing code from above
  void _startWebSocketListener() {
    final wsUrl = '$websocketUrl/workstream_summaries/stream/identifiers';
    _wsChannel = WebSocketChannel.connect(Uri.parse(wsUrl));

print('WebSocket connected to $wsUrl');

_wsSubscription = _wsChannel!.stream.listen(
      (message) async {
        try {
          // Important: WebSocket sends JSON strings, decode them first!
          final streamedIdentifiers = StreamedIdentifiers.fromJson(
            jsonDecode(message),
          );

// Loop through each identifier we received
          for (final id in streamedIdentifiers!.iterable) {
            final summaryId = id.workstreamSummary?.id;
            if (summaryId != null) {
              await _fetchAndCacheSummary(summaryId);
            }
          }
        } catch (e) {
          print('Error processing message: $e');
        }
      },
      onDone: () {
        print('WebSocket closed, reconnecting...');
        _reconnectWebSocket();
      },
      onError: (error) {
        print('WebSocket error: $error');
        _reconnectWebSocket();
      },
    );
  }

这种方式的妙处在于：WebSocket 会一直保持连接，直到应用关闭或用户手动断开，并持续向我们发送标识符。首次连接时，它会一次性发送所有现有标识符（甚至可能包括一年前的！），然后保持连接，在新标识符实时创建时继续发送给我们。只需一个连接就能处理所有事情！

第 3 步：获取并缓存摘要

好了，现在我们已经有了标识符。接下来呢？我们需要实际获取摘要数据并对其进行整理。

下面是 WorkstreamSummary 的结构（至少是其中的重要部分）：

// This is a model defined in the sdks don't add any thing to the code yet.

class WorkstreamSummary {
  String id;              // Unique identifier
  String? name;           // Optional name/title
  GroupedTimestamp created;  // When it was created
  GroupedTimestamp? updated; // When it was last updated
  // ... and a bunch of other fields
}

而 GroupedTimestamp 的基本结构如下：

// This is a model defined in the sdks don't add any thing to the code yet.

class GroupedTimestamp {
  DateTime value;  // The actual timestamp
}

因此，获取摘要时，我们需要：

去掉日期中的时间部分

将摘要添加到对应日期的列表中

按时间对所有内容排序

// existing code from above
  Future<void> _fetchAndCacheSummary(String identifier) async {
    final summary = await _workstreamSummaryApi
        .workstreamSummariesSpecificWorkstreamSummarySnapshot(identifier);

if (summary == null) {
      return;
    }
    // Normalize to day (remove time)
    final createdDate = summary.created.value;
    final dayKey = DateTime(
      createdDate.year,
      createdDate.month,
      createdDate.day,
    );

// Create day entry if it doesn't exist
    _summariesByDay.putIfAbsent(dayKey, () => []);

// Check for duplicates (avoid adding the same summary twice)
    final existingIndex = _summariesByDay[dayKey]!
        .indexWhere((s) => s.id == summary.id);

if (existingIndex != -1) {
      // Update existing
      _summariesByDay[dayKey]![existingIndex] = summary;
    } else {
      // Add new
      _summariesByDay[dayKey]!.add(summary);
    }

// Sort by time (most recent first)
    _summariesByDay[dayKey]!.sort((a, b) {
      return b.created.value.compareTo(a.created.value);
    });

// Notify anyone listening
    _summariesStreamController.add(Map.unmodifiable(_summariesByDay));

print('Cached summary ${summary.id} for day $dayKey');
  }

_summariesByDay 只是一个 Map<DateTime, List<WorkstreamSummary>>，其中：

键：设置为午夜的 DateTime（例如 2025 年 11 月 4 日 00:00:00）

值：当天所有摘要组成的列表，并按时间排序

我们再添加一些用于获取摘要的实用方法：

 // existing code
  /// Get summaries for a specific day from cache
  List<WorkstreamSummary> getSummariesForDay(DateTime date) {
    final dayKey = DateTime(date.year, date.month, date.day);
    return _summariesByDay[dayKey] ?? [];
  }

/// Get all days that have summaries (sorted most recent first)
  List<DateTime> getDaysWithSummaries() {
    final days = _summariesByDay.keys.toList();
    days.sort((a, b) => b.compareTo(a)); // Most recent first
    return days;
  }

第 4 步：保持所有内容同步

说到流——让我们添加一个可供其他组件监听的广播流：

// existing code
final StreamController<Map<DateTime, List<WorkstreamSummary>>>
    _summariesStreamController = StreamController.broadcast();

Stream<Map<DateTime, List<WorkstreamSummary>>> get summariesStream =>
    _summariesStreamController.stream;

这样一来，在 UI 中（等我们构建它时），只需这样做：

service.summariesStream.listen((summaries) {
  print('Got new data!');
  // Update UI here
});

第 4.5 步：等待 WebSocket 完成初始同步

这里有一个问题：应用首次启动时，WebSocket 会一次性向我们发送所有现有摘要。这可能包含数百个标识符！如果我们在它们全部加载完成之前就尝试显示 UI 或生成回顾，得到的数据将是不完整的。

但我们只需要等待一次——也就是第一条消息。之后，WebSocket 只会实时发送新的更新，我们不希望因此阻塞。

下面是使用 Completer 的解决方案：

将此代码块放在 PiecesOSService() 构造函数之后、connectApplication() 方法之前：

// Add these fields to the class
Completer<void>? _initialSyncCompleter;
bool get _isInitialSyncComplete =>
      _initialSyncCompleter?.isCompleted ?? false;
/// Wait for the initial WebSocket sync to complete
/// Only blocks on the first call, returns immediately after
Future<void> waitForInitialSync() async {
  // If already synced, return immediately
  if (_isInitialSyncComplete) {
    return;
  }

// If sync is in progress, wait for it
  if (_initialSyncCompleter != null) {
    return _initialSyncCompleter!.future;
  }

// Start waiting for first sync
  _initialSyncCompleter = Completer<void>();

替换 _wsSubscription 为这个新部分：

_wsSubscription = _wsChannel!.stream.listen(
  (message) async {
    try {
      final streamedIdentifiers = StreamedIdentifiers.fromJson(
        jsonDecode(message),
      );

// 处理所有标识符
      for (final id in streamedIdentifiers!.iterable) {
        final summaryId = id.workstreamSummary?.id;
        if (summaryId != null) {
          await _fetchAndCacheSummary(summaryId);
        }
      }

// 在收到第一条消息后标记初始同步完成
      if (!_isInitialSyncComplete) {
        _isInitialSyncComplete = true;
        _initialSyncCompleter?.complete();
        print('Initial WebSocket sync complete!');
      }
    } catch (e) {
      print('Error processing message: $e');
    }
  },
  // ... onDone、onError 处理器
);

在你应用的入口点（通常是 lib/main.dart，如果没有则创建该文件）放置这段代码：

import 'services/pieces_os_service.dart';

Future<void> main() async {
  final service = PiecesOSService();

await service.connectApplication();
  service.startWebSocketListener();

// 等待所有现有摘要加载完成
  await service.waitForInitialSync();

// 现在我们可以安全地生成回顾或显示 UI！
  print('Ready to go! All summaries loaded.');
}

Step 5：访问实际的摘要内容

来自 SDK 的 WorkstreamSummary 对象不会直接公开摘要文本。实际内容存储在注解中。

每个摘要都有注解，我们需要：

• 遍历摘要的注解
• 找到类型为 SUMMARY 的注解
• 使用 AnnotationApi 获取该注解
• 提取文本内容

让我们在 pieces_os_service.dart 底部添加一个方法来获取注解内容：

// 在 pieces_os_service.dart 底部

/// 从工作流摘要的注解中获取摘要内容
Future<String?> getSummaryContent(WorkstreamSummary summary) async {
  try {
    // 遍历注解以找到 DESCRIPTION 类型
    for (final annotationRef in summary.annotations?.indices.keys.toList() ?? []) {
      // 使用 AnnotationApi 获取完整注解
      final annotation = await _annotationApi
          .annotationSpecificAnnotationSnapshot(annotationRef);
 
      if (annotation == null) {
        continue;
      }
      // 检查这是否是 SUMMARY 类型注解
      if (annotation.type == AnnotationTypeEnum.SUMMARY) {
        // 返回文本内容 - 这就是实际的摘要！
        return annotation.text;
      }
    }

return null;
  } catch (e) {
    print('Error fetching annotation content for ${summary.id}: $e');
    return null;
  }
}

SDK 将元数据（ID、时间戳）与内容（存储在注解中）分离。

Step 6：创建适当的数据结构

让我们为包含内容的摘要创建一个合适的类。将此添加到 lib/models/daily_recap_models.dart：

/// 用于保存摘要及其内容的数据类
class SummaryWithContent {
  final String id;
  final String title;
  final String content;        // 实际的摘要文本！
  final DateTime timestamp;

SummaryWithContent({
    required this.id,
    required this.title,
    required this.content,
    required this.timestamp,
  });
}

现在，让我们添加一个方便的方法来获取包含内容的摘要：

// 别忘了导入我们刚创建的模型
import 'package:daily_recap_app/models/daily_recap_models.dart';

// pieces_os_service.dart 中现有代码下面

/// 获取特定日期的包含内容的摘要
Future<List<SummaryWithContent>> getSummariesWithContentForDay(
  DateTime date,
) async {
  final summaries = getSummariesForDay(date);
  final List<SummaryWithContent> summariesWithContent = [];

for (final summary in summaries) {
    final content = await getSummaryContent(summary);
    if (content != null && content.isNotEmpty) {
      summariesWithContent.add(
        SummaryWithContent(
          id: summary.id,
          title: summary.name,
          content: content,              // 来自注解的实际文本！
          timestamp: summary.created.value,
        ),
      );
    }
  }

return summariesWithContent;
}

完美！现在我们拥有已准备好用于 AI 处理和 UI 显示的丰富、类型化的数据。

• 打开 WebSocket 进行实时更新
• 获取并缓存摘要
• 从注解中提取摘要内容

迄今为止创建的内容：lib/services/pieces_os_service.dart

import 'dart:async';
import 'dart:convert';
import 'dart:io';
import 'package:pieces_os_client/api.dart';
import 'package:pieces_os_client/api_client.dart';
import 'package:web_socket_channel/web_socket_channel.dart';
import 'package:daily_recap_app/models/daily_recap_models.dart';

/// 与 Pieces OS 交互以进行 LTM（长期记忆）跟踪的服务
class PiecesOSService {
  static const String baseUrl = 'http://localhost:39300';
  static const String websocketUrl = 'ws://localhost:39300';

// 用于注册应用程序
  late final ConnectorApi _connectorApi;
  // 用于获取工作流摘要
  late final WorkstreamSummaryApi _workstreamSummaryApi;

// 用于获取摘要内容
  late final AnnotationApi _annotationApi;

final StreamController<Map<DateTime, List<WorkstreamSummary>>>
      _summariesStreamController = StreamController.broadcast();

Stream<Map<DateTime, List<WorkstreamSummary>>> get summariesStream =>
      _summariesStreamController.stream;

// 跟踪初始 WebSocket 同步
  bool _isInitialSyncComplete = false;
  Completer<void>? _initialSyncCompleter;

PiecesOSService() {
    final client = ApiClient(basePath: baseUrl);
    _connectorApi = ConnectorApi(client);
    _workstreamSummaryApi = WorkstreamSummaryApi(client);
    _annotationApi = AnnotationApi(client);
  }

Future<Application> connectApplication() async {
    final seededApp = SeededTrackedApplication(
      name: ApplicationNameEnum.OPEN_SOURCE,  // 嘿 Pieces，我是开源的！
      platform: Platform.operatingSystem == "macos"
          ? PlatformEnum.MACOS
          : PlatformEnum.WINDOWS,
      version: "0.0.1",
    );

final connection = SeededConnectorConnection(
      application: seededApp,
    );

_context = await _connectorApi.connect(
      seededConnectorConnection: connection,
    );

print('Successfully connected to Pieces OS!');
    return _context!.application;
  }

/// 等待初始 WebSocket 同步完成
  /// 仅在第一次调用时阻塞，之后立即返回
  Future<void> waitForInitialSync() async {
    // 如果已同步，立即返回
    if (_isInitialSyncComplete) {
      return;
    }

// 如果同步进行中，等待它
    if (_initialSyncCompleter != null) {
      return _initialSyncCompleter!.future;
    }

// 开始等待第一次同步
    _initialSyncCompleter = Completer<void>();
    return _initialSyncCompleter!.future;
  }

void _startWebSocketListener() {
    final wsUrl = '$websocketUrl/workstream_summaries/stream/identifiers';
    _wsChannel = WebSocketChannel.connect(Uri.parse(wsUrl));

print('WebSocket connected to $wsUrl');

_wsSubscription = _wsChannel!.stream.listen(
      (message) async {
        try {
          // 重要：WebSocket 发送 JSON 字符串，首先解码！
          final streamedIdentifiers = StreamedIdentifiers.fromJson(
            jsonDecode(message),
          );

// 遍历我们收到的每个标识符
          for (final id in streamedIdentifiers!.iterable) {
            final summaryId = id.workstreamSummary?.id;
            if (summaryId != null) {
              await _fetchAndCacheSummary(summaryId);
            }
          }

// 在收到第一条消息后标记初始同步完成
          if (!_isInitialSyncComplete) {
            _isInitialSyncComplete = true;
            _initialSyncCompleter?.complete();
            print('Initial WebSocket sync complete!');
          }
        } catch (e) {
          print('Error processing message: $e');
        }
      },
      onDone: () {
        print('WebSocket closed, reconnecting...');
        _reconnectWebSocket();
      },
      onError: (error) {
        print('WebSocket error: $error');
        _reconnectWebSocket();
      },
    );
  }

Future<void> _fetchAndCacheSummary(String identifier) async {
    final summary = await _workstreamSummaryApi
        .workstreamSummariesSpecificWorkstreamSumm

查看 GitHub 以查看完整项目。