@earendil-works / pi-agent-core

具有工具执行和事件流的有状态代理。建立在 @earendil-works/pi-ai.

安装

npm install @earendil-works/pi-agent-core

# Bun alternative
bun add @earendil-works/pi-agent-core

SQLite会话后端

SQLite会话后端和 node:sqlite 适配器位于一个单独的包中， @earendil-works/pi-storage-sqlite-node，因此core包默认情况下不会拉入运行时内置或本地SQLite依赖项。后端接受特定于运行时的SQLite工厂，允许其他会话后端在将来作为自己的包发布。

代理利用

这个叉保持代理的线束(src/harness/**)，连同它的子路径导出(./node， ./harness/*， ./experimental/pico3)和SQLite会话后端包。Upstream v1.0.0将该运行时移到了 pi-durable;分叉没有采用这个动作，所以coding-agent继续从 @earendil-works/pi-agent-core.

快速入门

import { Agent } from "@earendil-works/pi-agent-core";
import { createModels } from "@earendil-works/pi-ai";
import { anthropicProvider } from "@earendil-works/pi-ai/providers/anthropic";

const models = createModels();
models.setProvider(anthropicProvider());
const model = models.getModel("anthropic", "claude-sonnet-4-6");
if (!model) throw new Error("Model not found");

const agent = new Agent({
  initialState: {
    systemPrompt: "You are a helpful assistant.",
    model,
  },
  streamFn: models.streamSimple.bind(models),
});

agent.subscribe((event) => {
  if (event.type === "message_update" && event.assistantMessageEvent.type === "text_delta") {
    // Stream just the new text chunk
    process.stdout.write(event.assistantMessageEvent.delta);
  }
});

await agent.prompt("Hello!");

实验性面服务

传输中立的面服务原语驻留在 @earendil-works/chord． 代理核心不导出服务运行时。

核心概念

AgentMessage vs LLM Message

代理与 AgentMessage，一种灵活的类型，可以包括：

• 标准LLM消息(user， assistant， toolResult）
• 通过声明合并自定义应用特定的消息类型

法学硕士只懂 user， assistant, toolResult。这个 convertToLlm 函数通过在每次LLM调用之前过滤和转换消息来弥补这一差距。

消息流

AgentMessage[] → transformContext() → AgentMessage[] → convertToLlm() → Message[] → LLM
                    (optional)                           (required)

1. transformContext：删减旧信息，注入外部背景
2. convertToLlm：过滤掉纯ui消息，将自定义类型转换为LLM格式

事件流

代理为UI更新发出事件。理解事件序列有助于构建响应式接口。

prompt（）事件顺序

当你打来电话 prompt("Hello")：

prompt("Hello")
├─ agent_start
├─ turn_start
├─ message_start   { message: userMessage }      // Your prompt
├─ message_end     { message: userMessage }
├─ message_start   { message: assistantMessage } // LLM starts responding
├─ message_update  { message: partial... }       // Streaming chunks
├─ message_update  { message: partial... }
├─ message_end     { message: assistantMessage } // Complete response
├─ turn_end        { message, toolResults: [] }
└─ agent_end       { messages: [...] }

使用工具调用

如果助手调用工具，循环继续：

prompt("Read config.json")
├─ agent_start
├─ turn_start
├─ message_start/end  { userMessage }
├─ message_start      { assistantMessage with toolCall }
├─ message_update...
├─ message_end        { assistantMessage }
├─ tool_execution_start  { toolCallId, toolName, args }
├─ tool_execution_update { partialResult }           // If tool streams
├─ tool_execution_end    { toolCallId, result }
├─ message_start/end  { toolResultMessage }
├─ turn_end           { message, toolResults: [toolResult] }
│
├─ turn_start                                        // Next turn
├─ message_start      { assistantMessage }           // LLM responds to tool result
├─ message_update...
├─ message_end
├─ turn_end
└─ agent_end

工具执行模式可配置：

• parallel （默认）：按顺序调用预飞行工具，并发执行允许的工具，发出 tool_execution_end 一旦每个工具完成，然后发出toolResult消息和 turn_end.toolResults 在辅助源订单中
• sequential：逐个执行工具调用，匹配历史行为

在并行模式中，工具完成事件遵循工具完成顺序，但是持久化的toolResult消息仍然遵循辅助源顺序。

可以全局设置该模式 toolExecution 在代理配置中，或在每个工具中 executionMode 在 AgentTool． 在并行模式中，针对工具的工具调用 executionMode: "sequential" 作为排他性屏障运行：它等待批处理中所有先前计划的调用完成，之后的调用等待它完成。在屏障之前或之后的并行工具仍然彼此并发地执行。

这个 beforeToolCall 钩跑后 tool_execution_start 和有效的参数解析。它可以阻止执行和附加 terminate: true 到阻塞的结果。这个 afterToolCall 钩子在工具执行完成之后和之前运行 tool_execution_end 并发出最终的工具结果消息事件。

工具,阻塞 beforeToolCall 结果, afterToolCall 重写可以返回 terminate: true 提示应该跳过自动后续LLM调用。只有当每个最终确定的工具产生该批集时，循环才会提前停止 terminate: true． 混合批次继续正常运行。

当你使用 Agent 助理类 message_end 在刀具预飞开始之前，加工被视为一个障碍。这意味着 beforeToolCall 查看已包含请求工具调用的助手消息的代理状态。

申请准备和终场

prepareRequest 在每个会话提供程序请求（包括第一个）之前立即运行。在发出挂起输入后，使用它来安装规范持久化上下文：

agent.prepareRequest = async ({ context }) => ({
  context: { ...context, messages: await session.loadModelContext() },
});

prepareRequest 不轮询队列。转向队列在运行时等待下一个正常的转向轮询。

finishTurn 在辅助消息和所有工具结果最终确定之后运行 turn_end． 它可以运行正常、错误和中止的响应：

agent.finishTurn = async ({ message }) => {
  if (message.stopReason === "error" || message.stopReason === "aborted") return;
  if (shouldEndRun(message)) return { action: "end" };
  return needsAnotherResponse(message) ? { action: "continue" } : undefined;
};

返回 undefined 保持正常的调度。 { action: "end" } 之后就停止了 turn_end，然后轮询转向或后续队列或准备另一个请求。在正常反应中， { action: "continue" } 确保再发生一个提供者请求：工具结果、转向或已经引起该请求的后续操作满足该请求，否则循环只发出一个上下文请求。错误和中止的响应仍然是硬出口，因此它们的决定被忽略。 finishTurn 在下一个请求后再次运行，因此返回 { action: "continue" } 无条件循环永远。低级循环调用者设置了相同的钩子 AgentLoopConfig.

迁移：从被移走的地方迁移 shouldStopAfterTurn,还 { action: "end" } 并保护错误和中止的响应，以保持旧的normal-response-only调用：

finishTurn: async (turn, signal) => {
  if (turn.message.stopReason === "error" || turn.message.stopReason === "aborted") return;
  return (await shouldStop(turn, signal)) ? { action: "end" } : undefined;
},

每个提供商都遵循这样的生命周期：

selected input events
→ prepareRequest
→ provider response
→ tool results
→ finishTurn
→ turn_end
→ existing continuation scheduling or agent_end

agent.peekQueuedMessages() 根据队列模式预览下一个队列选择的批处理，而不使用它。

continue（）事件顺序

continue() 从现有上下文中恢复，而不添加新消息。错误后重试时使用它。

// After an error, retry from current state
await agent.continue();

// Keep queued user input out of the retry request and use shorter liveness
// bounds without mutating the Agent's configured defaults.
await agent.continue({
  deferQueuedMessages: true,
  timeoutMs: 30_000,
  streamStartTimeoutMs: 30_000,
});

上下文中的最后一条消息必须是 user 或者 toolResult (不 assistant）. 继续选项有 一次性：队列延迟和两个超时覆盖仅应用于第一个提供者请求。如果这 请求成功，并通过工具或排队输入继续相同的运行，稍后提供程序请求使用 代理已配置 timeoutMs 和 streamStartTimeoutMs. continueWithQueuedMessages(options) 保持压缩恢复队列优先；所选的排队消息将成为延续输入，因此 deferQueuedMessages 当第一次请求超时覆盖仍然适用时，不能延迟该消息。

事件类型

Agent.subscribe() 监听器按注册顺序等待。 agent_end 意味着不会再发出循环事件，但是 await agent.waitForIdle() 和 await agent.prompt(...) 等待过后才安定下来 agent_end 侦听器完成。

剂的选择

const agent = new Agent({
  // Initial state
  initialState: {
    systemPrompt: string,
    model: Model<any>,
    thinkingLevel: "off" | "minimal" | "low" | "medium" | "high" | "xhigh" | "max",
    tools: AgentTool<any>[],
    messages: AgentMessage[],
  },

  // Convert AgentMessage[] to LLM Message[] (required for custom message types)
  convertToLlm: (messages) => messages.filter(...),

  // Transform context before convertToLlm (for pruning, compaction)
  transformContext: async (messages, signal) => pruneOldMessages(messages),

  // Steering mode: "one-at-a-time" (default) or "all"
  steeringMode: "one-at-a-time",

  // Follow-up mode: "one-at-a-time" (default) or "all"
  followUpMode: "one-at-a-time",

  // Required stream function
  streamFn: models.streamSimple.bind(models),

  // Session ID for provider caching
  sessionId: "session-123",

  // Dynamic API key resolution (for expiring OAuth tokens)
  getApiKey: async (provider) => refreshToken(),

  // Tool execution mode: "parallel" (default) or "sequential"
  toolExecution: "parallel",

  // Preflight each tool call after args are validated. Can block execution.
  beforeToolCall: async ({ toolCall, args, context }) => {
    if (toolCall.name === "bash") {
      return { block: true, reason: "bash is disabled", terminate: true };
    }
  },

  // Postprocess each tool result before final tool events are emitted.
  afterToolCall: async ({ toolCall, result, isError, context }) => {
    if (toolCall.name === "notify_done" && !isError) {
      return { terminate: true };
    }
    if (!isError) {
      return { details: { ...result.details, audited: true } };
    }
  },

  // End the run gracefully after a completed turn, before queued messages are polled.
  finishTurn: async ({ message, context }, signal) => {
    if (message.stopReason === "error" || message.stopReason === "aborted") return;
    return (await shouldCompactBeforeNextTurn(context.messages, signal)) ? { action: "end" } : undefined;
  },

  // Custom thinking budgets for token-based providers
  thinkingBudgets: {
    minimal: 128,
    low: 512,
    medium: 1024,
    high: 2048,
  },
});

代理国家

interface AgentState {
  systemPrompt: string;
  model: Model<any>;
  thinkingLevel: ThinkingLevel;
  tools: AgentTool<any>[];
  messages: AgentMessage[];
  readonly isStreaming: boolean;
  readonly streamingMessage?: AgentMessage;
  readonly pendingToolCalls: ReadonlySet<string>;
  readonly errorMessage?: string;
}

通过访问状态 agent.state.

分配 agent.state.tools = [...] 或者 agent.state.messages = [...] 在存储顶级数组之前复制它。改变返回的数组会改变当前代理状态。

在流, agent.state.streamingMessage 包含当前的部分助手消息。

agent.state.isStreaming 仍然是 true 直到跑步完全结束，包括等待 agent_end 订阅者。

方法

促使

// Text prompt
await agent.prompt("Hello");

// With images
await agent.prompt("What's in this image?", [
  { type: "image", data: base64Data, mimeType: "image/jpeg" }
]);

// AgentMessage directly
await agent.prompt({ role: "user", content: "Hello", timestamp: Date.now() });

// Continue from current context (last message must be user or toolResult)
await agent.continue();

状态管理

agent.state.systemPrompt = "New prompt";
agent.state.model = getModel("openai", "gpt-4o");
agent.state.thinkingLevel = "medium";
agent.state.tools = [myTool];
agent.toolExecution = "sequential";
agent.beforeToolCall = async ({ toolCall }) => undefined;
agent.afterToolCall = async ({ toolCall, result }) => undefined;
agent.prepareRequest = async ({ context }) => undefined;
agent.finishTurn = async () => undefined;
agent.state.messages = newMessages; // top-level array is copied
agent.state.messages.push(message);
agent.reset();

会议和思考预算

agent.sessionId = "session-123";

agent.thinkingBudgets = {
  minimal: 128,
  low: 512,
  medium: 1024,
  high: 2048,
};

控制

agent.abort();           // Cancel current operation
await agent.waitForIdle(); // Wait for completion

事件

const unsubscribe = agent.subscribe(async (event, signal) => {
  if (event.type === "agent_end") {
    // Final barrier work for the run
    await flushSessionState(signal);
  }
});
unsubscribe();

督导及跟进

控制消息允许您在工具运行时中断代理。后续消息允许您在代理停止后对工作进行排队。

agent.steeringMode = "one-at-a-time";
agent.followUpMode = "one-at-a-time";

// While agent is running tools
agent.steer({
  role: "user",
  content: "Stop! Do this instead.",
  timestamp: Date.now(),
});

// After the agent finishes its current work
agent.followUp({
  role: "user",
  content: "Also summarize the result.",
  timestamp: Date.now(),
});

const steeringMode = agent.steeringMode;
const followUpMode = agent.followUpMode;

agent.clearSteeringQueue();
agent.clearFollowUpQueue();
agent.clearAllQueues();

使用clearSteeringQueue， clearFollowUpQueue，或clearAllQueues来删除排队的消息。

当转弯完成后检测到转向信息时：

1. 来自当前助手消息的所有工具调用已经完成
2. 注入转向信息
3. LLM在下一个转弯时做出反应

只有在没有更多的工具调用和转向消息时才检查后续消息。如果有队列，则注入它们并运行另一个回合。

自定义消息类型

扩展 AgentMessage 通过声明合并：

declare module "@earendil-works/pi-agent-core" {
  interface CustomAgentMessages {
    notification: { role: "notification"; text: string; timestamp: number };
  }
}

// Now valid
const msg: AgentMessage = { role: "notification", text: "Info", timestamp: Date.now() };

中处理自定义类型 convertToLlm：

const agent = new Agent({
  streamFn: models.streamSimple.bind(models),
  convertToLlm: (messages) => messages.flatMap(m => {
    if (m.role === "notification") return []; // Filter out
    return [m];
  }),
});

工具

定义工具 AgentTool：

import { Type } from "typebox";

const readFileTool: AgentTool = {
  name: "read_file",
  label: "Read File",  // For UI display
  description: "Read a file's contents",
  parameters: Type.Object({
    path: Type.String({ description: "File path" }),
  }),
  // Override execution mode for this tool (optional).
  // "sequential" makes this tool an exclusive barrier in parallel batches:
  // it runs alone, but parallel tools before or after it still run concurrently.
  // "parallel" allows concurrent execution with other tool calls.
  // If omitted, the global toolExecution config applies.
  executionMode: "sequential",
  execute: async (toolCallId, params, signal, onUpdate) => {
    const content = await fs.readFile(params.path, "utf-8");

    // Optional: stream progress
    onUpdate?.({ content: [{ type: "text", text: "Reading..." }], details: {} });

    // Optional: add `terminate: true` here to skip the automatic follow-up LLM call
    // when every finalized tool result in the batch does the same.
    return {
      content: [{ type: "text", text: content }],
      details: { path: params.path, size: content.length },
    };
  },
};

agent.state.tools = [readFileTool];

错误处理

抛出错误 当工具出现故障时。不要将错误消息作为内容返回。

execute: async (toolCallId, params, signal, onUpdate) => {
  if (!fs.existsSync(params.path)) {
    throw new Error(`File not found: ${params.path}`);
  }
  // Return content only on success
  return { content: [{ type: "text", text: "..." }] };
}

抛出的错误被代理捕获，并作为工具错误报告给LLM isError: true.

返回 terminate: true 从 execute()，阻塞 beforeToolCall，或者 afterToolCall 提示代理应在当前刀具批处理后停止。这仅在批处理中每个最终确定的工具结果终止时生效。提示是仅运行时的；发出 toolResult 转录信息仍然是标准的LLM工具结果。

代理的使用

对于通过后端代理的浏览器应用程序：

import { Agent, streamProxy } from "@earendil-works/pi-agent-core";

const agent = new Agent({
  streamFn: (model, context, options) =>
    streamProxy(model, context, {
      ...options,
      authToken: "...",
      proxyUrl: "https://your-server.com",
    }),
});

低级API

对于不使用Agent类的直接控制：

import { agentLoop, agentLoopContinue } from "@earendil-works/pi-agent-core";

const context: AgentContext = {
  systemPrompt: "You are helpful.",
  messages: [],
  tools: [],
};

const config: AgentLoopConfig = {
  model: getModel("openai", "gpt-4o"),
  convertToLlm: (msgs) => msgs.filter(m => ["user", "assistant", "toolResult"].includes(m.role)),
  toolExecution: "parallel",  // overridden by per-tool executionMode if set
  beforeToolCall: async ({ toolCall, args, context }) => undefined,
  afterToolCall: async ({ toolCall, result, isError, context }) => undefined,
};

const userMessage = { role: "user", content: "Hello", timestamp: Date.now() };

const streamFn = models.streamSimple.bind(models);
for await (const event of agentLoop([userMessage], context, config, undefined, streamFn)) {
  console.log(event.type);
}

// Continue from existing context
for await (const event of agentLoopContinue(context, config, undefined, streamFn)) {
  console.log(event.type);
}

这些低空气流是可观测的。它们保留事件顺序，但它们不会等待异步事件处理完成，然后才继续后面的生产者阶段。如果您需要消息处理在工具预飞行之前充当屏障，请使用 Agent Class代替raw agentLoop() 或者 agentLoopContinue().

许可证

麻省理工学院