22  JSON 事件流模式

22.1 JSON 事件流模式概述

JSON 事件流模式是 Pi 最简单的编程式集成方式。它会将会话中的所有事件以 JSON 行（JSONL）格式输出到 stdout，无需任何协议交互——只需发送一个提示，然后解析输出即可。

pi --mode json "Your prompt"

这种模式特别适合：

• 将 Pi 集成到 CI/CD 流水线
• 构建简单的自动化脚本
• 调试 Agent 行为
• 与其他工具进行数据交换

Tip

相比 RPC 模式，JSON 模式更简单：不需要双向通信，只需要单次提示 + 事件流解析。

22.2 基本用法

# 基本用法
pi --mode json "What files are in the current directory?"

# 指定模型
pi --mode json --model anthropic/claude-sonnet-4-5 "Explain this code"

# 指定提供者和思考级别
pi --mode json --model anthropic/claude-opus-4-5:high "Design a system architecture"

# 带图片
pi --mode json --image /path/to/image.png "What's in this image?"

22.3 输出格式

每行是一个 JSON 对象。第一行是会话头部：

{"type":"session","version":3,"id":"uuid","timestamp":"2026-07-31T12:00:00Z","cwd":"/path"}

随后是按顺序发生的事件：

{"type":"agent_start"}
{"type":"turn_start"}
{"type":"message_start","message":{"role":"assistant","content":[],"model":"claude-sonnet-4-5"}}
{"type":"message_update","message":{...},"assistantMessageEvent":{"type":"text_delta","delta":"Hello"}}
{"type":"message_update","message":{...},"assistantMessageEvent":{"type":"text_delta","delta":", "}}
{"type":"message_update","message":{...},"assistantMessageEvent":{"type":"text_delta","delta":"world!"}}
{"type":"message_end","message":{...}}
{"type":"turn_end","message":{...},"toolResults":[]}
{"type":"agent_end","messages":[...]}

22.4 事件类型

22.4.1 AgentSessionEvent 类型定义

type AgentSessionEvent =
  | AgentEvent
  | { type: "queue_update"; steering: readonly string[]; followUp: readonly string[] }
  | { type: "compaction_start"; reason: "manual" | "threshold" | "overflow" }
  | { type: "compaction_end"; reason: "manual" | "threshold" | "overflow"; result: CompactionResult | undefined; aborted: boolean; willRetry: boolean; errorMessage?: string }
  | { type: "auto_retry_start"; attempt: number; maxAttempts: number; delayMs: number; errorMessage: string }
  | { type: "auto_retry_end"; success: boolean; attempt: number; finalError?: string }
  | { type: "summarization_retry_scheduled"; attempt: number; maxAttempts: number; delayMs: number; errorMessage: string }
  | { type: "summarization_retry_attempt_start"; source: "branchSummary" }
  | { type: "summarization_retry_attempt_start"; source: "compaction"; reason: "manual" | "threshold" | "overflow" }
  | { type: "summarization_retry_finished" };

22.4.2 AgentEvent 基础事件

type AgentEvent =
  // Agent 生命周期
  | { type: "agent_start" }
  | { type: "agent_end"; messages: AgentMessage[] }
  // 回合生命周期
  | { type: "turn_start" }
  | { type: "turn_end"; message: AgentMessage; toolResults: ToolResultMessage[] }
  // 消息生命周期
  | { type: "message_start"; message: AgentMessage }
  | { type: "message_update"; message: AgentMessage; assistantMessageEvent: AssistantMessageEvent }
  | { type: "message_end"; message: AgentMessage }
  // 工具执行
  | { type: "tool_execution_start"; toolCallId: string; toolName: string; args: any }
  | { type: "tool_execution_update"; toolCallId: string; toolName: string; args: any; partialResult: any }
  | { type: "tool_execution_end"; toolCallId: string; toolName: string; result: any; isError: boolean };

22.4.3 queue_update 详解

queue_update 事件在待处理队列变更时发出，包含完整的待处理 steering 和 follow-up 队列：

{
  "type": "queue_update",
  "steering": ["First steering message"],
  "followUp": ["First follow-up message", "Second follow-up message"]
}

22.4.4 compaction 事件详解

compaction_start 和 compaction_end 覆盖手动和自动压缩：

{"type": "compaction_start", "reason": "threshold"}
{"type": "compaction_end", "reason": "threshold", "result": {...}, "aborted": false, "willRetry": false}

22.5 消息类型

22.5.1 基础消息

来自 @earendil-works/pi-ai：

类型	说明
UserMessage	用户发送的消息
AssistantMessage	助手响应消息
ToolResultMessage	工具执行结果

22.5.2 扩展消息

来自 Pi 编程助手核心：

类型	说明
BashExecutionMessage	bash 命令执行记录
CustomMessage	扩展自定义消息
BranchSummaryMessage	分支摘要
CompactionSummaryMessage	压缩摘要

22.6 使用示例

22.6.1 提取最终回复文本

# 使用 jq 提取所有文本增量
pi --mode json "Explain async/await" 2>/dev/null | \
  jq -c 'select(.type == "message_update") | .assistantMessageEvent | select(.type == "text_delta") | .delta' | \
  tr -d '"' | tr -d '\n'

22.6.2 提取完整消息

# 只输出 message_end 事件
pi --mode json "List files" 2>/dev/null | jq -c 'select(.type == "message_end")'

22.6.3 监控工具执行

# 显示工具调用
pi --mode json "Fix the bug in main.ts" 2>/dev/null | \
  jq -c 'select(.type == "tool_execution_start") | {tool: .toolName, args: .args}'

22.6.4 统计 Token 使用量

# 提取 agent_end 中的消息统计
pi --mode json "Analyze code quality" 2>/dev/null | \
  jq -c 'select(.type == "agent_end") | {messages: (.messages | length)}'

22.6.5 Python 集成示例

import subprocess
import json

def run_pi(prompt: str) -> list[dict]:
    """运行 Pi JSON 模式，返回事件列表"""
    result = subprocess.run(
        ["pi", "--mode", "json", prompt],
        capture_output=True,
        text=True,
    )
    events = []
    for line in result.stdout.strip().split("\n"):
        if line:
            events.append(json.loads(line))
    return events

def extract_text(events: list[dict]) -> str:
    """从事件流中提取助手回复文本"""
    text = ""
    for event in events:
        if event.get("type") == "message_update":
            ae = event.get("assistantMessageEvent", {})
            if ae.get("type") == "text_delta":
                text += ae["delta"]
    return text

# 使用
events = run_pi("What files are in the current directory?")
text = extract_text(events)
print(text)

22.6.6 Node.js 集成示例

import { spawn } from "child_process";

function runPi(prompt: string): Promise<string[]> {
  return new Promise((resolve, reject) => {
    const pi = spawn("pi", ["--mode", "json", prompt], {
      stdio: ["inherit", "pipe", "inherit"],
    });

    const lines: string[] = [];
    let buffer = "";

    pi.stdout?.on("data", (data) => {
      buffer += data.toString();
      const parts = buffer.split("\n");
      buffer = parts.pop() || "";
      lines.push(...parts.filter(l => l.trim()));
    });

    pi.on("close", (code) => {
      if (code !== 0) {
        reject(new Error(`Pi exited with code ${code}`));
      } else {
        resolve(lines.map(l => JSON.parse(l)));
      }
    });
  });
}

// 使用
async function main() {
  const events = await runPi("List all TypeScript files");

  let text = "";
  for (const event of events) {
    if (event.type === "message_update") {
      const ae = event.assistantMessageEvent;
      if (ae?.type === "text_delta") {
        text += ae.delta;
      }
    }
  }

  console.log(text);
}

main();

22.6.7 CI/CD 集成示例

#!/bin/bash
# 在 CI 中自动生成 commit message

set -e

# 获取 staged 变更
DIFF=$(git diff --cached --stat)

if [ -z "$DIFF" ]; then
  echo "No staged changes"
  exit 0
fi

# 使用 Pi 生成 commit message
COMMIT_MSG=$(pi --mode json "Generate a conventional commit message for these staged changes: $DIFF" 2>/dev/null | \
  jq -r 'select(.type == "message_update") | .assistantMessageEvent | select(.type == "text_delta") | .delta' | \
  tr -d '\n')

echo "Generated commit message:"
echo "$COMMIT_MSG"

# 自动提交
# git commit -m "$COMMIT_MSG"

22.7 JSON 模式 vs RPC 模式

特性	JSON 模式	RPC 模式
通信方式	单向输出	双向（stdin/stdout）
交互	单次提示	多轮对话
复杂度	简单	中等
适用场景	自动化、CI/CD	IDE 集成、自定义 UI
事件类型	相同	相同
扩展 UI	不支持	支持

Tip

如果只需要运行一次提示并获取结果，JSON 模式是最佳选择。如果需要多轮交互或扩展 UI 支持，使用 RPC 模式。