pi-agent-core 最小 Agent 运行时

import { Agent, type AgentTool } from "@earendil-works/pi-agent-core";
import {
  createModels,
  fauxAssistantMessage,
  fauxProvider,
  fauxText,
  fauxToolCall,
  Type,
} from "@earendil-works/pi-ai";

// 1) 用 faux provider 代替真实模型，脚本化地返回两轮响应（不需要 API Key）
const faux = fauxProvider();
const models = createModels();
models.setProvider(faux.provider);
const model = faux.getModel();

faux.setResponses([
  fauxAssistantMessage([fauxText("我先查一下天气。"), fauxToolCall("get_weather", { city: "北京" })], {
    stopReason: "toolUse",
  }),
  fauxAssistantMessage([fauxText("北京今天晴，25°C，很适合跑步。")]),
]);

// 2) 定义工具：TypeBox schema 同时负责类型推导和运行时校验
const weatherParams = Type.Object({ city: Type.String({ description: "城市名，例如 北京" }) });
const getWeather: AgentTool<typeof weatherParams> = {
  name: "get_weather",
  label: "查询天气",
  description: "查询城市今天的天气",
  parameters: weatherParams,
  execute: async (_toolCallId, params) => {
    const data: Record<string, string> = { 北京: "晴，25°C", 上海: "小雨，22°C" };
    return { content: [{ type: "text", text: data[params.city] ?? "暂无数据" }], details: { city: params.city } };
  },
};

// 3) 创建 Agent，并订阅事件
const agent = new Agent({
  initialState: { systemPrompt: "你是天气助手。", model, tools: [getWeather] },
  streamFn: models.streamSimple.bind(models),
});

agent.subscribe((event) => {
  if (event.type === "turn_start") console.log("── turn_start");
  if (event.type === "tool_execution_start") console.log(`   tool_execution_start: ${event.toolName}`, event.args);
  if (event.type === "tool_execution_end") console.log(`   tool_execution_end: isError=${event.isError}`);
  if (event.type === "message_end") console.log(`   message_end: role=${event.message.role}`);
  if (event.type === "agent_end") console.log(`── agent_end: 共 ${event.messages.length} 条新消息`);
});

await agent.prompt("北京今天适合跑步吗？");
const last = agent.state.messages.at(-1);
if (last?.role === "assistant") {
  console.log("最终回答：", last.content.filter((b) => b.type === "text").map((b) => b.text).join(""));
}

── turn_start
   message_end: role=user
   message_end: role=assistant
   tool_execution_start: get_weather { city: '北京' }
   tool_execution_end: isError=false
   message_end: role=toolResult
── turn_start
   message_end: role=assistant
── agent_end: 共 4 条新消息
最终回答： 北京今天晴，25°C，很适合跑步。

import { Agent } from "@earendil-works/pi-agent-core";
import { createModels, type Message } from "@earendil-works/pi-ai";

// 通过声明合并扩展 AgentMessage，加入一个只在 UI 里显示的 notification 类型
declare module "@earendil-works/pi-agent-core" {
  interface CustomAgentMessages {
    notification: { role: "notification"; text: string; timestamp: number };
  }
}

// 只有这几种角色是 LLM 能理解的，其余的（自定义类型）都要过滤或转换
const LLM_ROLES = new Set(["system", "user", "assistant", "toolResult"]);

const models = createModels();
const agent = new Agent({
  streamFn: models.streamSimple.bind(models),
  // 发给 LLM 之前，过滤掉 notification 这类自定义消息
  convertToLlm: (messages) => messages.filter((m): m is Message => LLM_ROLES.has(m.role)),
  // 每次请求前只保留最近 50 条消息，这是最朴素的上下文管理
  transformContext: async (messages) => messages.slice(-50),
});

agent.state.messages.push({ role: "notification", text: "已连接", timestamp: Date.now() });

// 伪代码：根据 agent-loop.ts 中 runLoop 的结构简化而来
pending = 取出 steering 消息（用户可能在等待时已经输入了内容）
while (true):                                   // 外层循环：处理 follow-up
    hasMoreToolCalls = true
    while (hasMoreToolCalls || pending 非空):   // 内层循环：处理工具调用和 steering
        把 pending 消息追加到上下文（并声明工具集的变化）
        prepareRequest?()                       // 每次请求前都可以重建上下文
        message = 流式请求 assistant 回复
        if message.stopReason in (error, aborted):
            finishTurn?(); emit turn_end; emit agent_end; return
        toolCalls = message 中的工具调用
        if toolCalls 非空:
            if message.stopReason == "length":  // 输出被截断，参数可能不完整
                把这批工具调用全部判为失败
            else:
                执行工具（并行或串行，带 before/after hook）
            hasMoreToolCalls = !(这批结果全部要求 terminate)
        decision = finishTurn?()                // 可以返回 end 或 continue
        emit turn_end
        if decision == end: emit agent_end; return
        pending = 取出 steering 消息
    followUps = 取出 follow-up 消息             // agent 原本要停了，看看还有没有后续任务
    if followUps 非空: pending = followUps; continue
    if 显式要求 continue: continue
    break
emit agent_end

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

const models = createModels();
const agent = new Agent({
  streamFn: models.streamSimple.bind(models),
  toolExecution: "parallel",
  // 预检：参数已经校验通过之后执行，可以拦截这次调用
  beforeToolCall: async ({ toolCall }) => {
    if (toolCall.name === "bash") {
      return { block: true, reason: "bash 已被禁用" };
    }
    return undefined;
  },
  // 后处理：在发出最终的工具事件之前，可以改写结果
  afterToolCall: async ({ toolCall, result, isError }) => {
    if (!isError && toolCall.name === "notify_done") {
      return { terminate: true }; // 提示：这批工具执行完后不必再请求 LLM
    }
    return undefined;
  },
});

// 在工具执行期间插话（steering），或者在当前任务完成后再加一个任务（follow-up）
agent.steer({ role: "user", content: "停一下，改为只检查 src 目录。", timestamp: Date.now() });
agent.followUp({ role: "user", content: "完成后顺便总结一下改动。", timestamp: Date.now() });