> Agent Harness Patterns 第 16 关，一条讲解 AI 智能体工作原理的模式路线。网页版：https://harnesspatterns.dev/zh/patterns/proactive-agents · 全部模式（英文）：https://harnesspatterns.dev/llms.txt

# 主动式智能体

到目前为止的每个智能体，都在等有人打字。主动式智能体会靠定时器自己醒来，四处看看，只在有值得说的事情时才开口。

## 问题

有些工作根本不存在一个让人想起来要问的时机：主人在学校时饿了的宠物、卡住的订单、半夜开始出故障的服务器。一个只有被叫到才回应的智能体，对这些情况毫无用处。

显而易见的办法，是用一个定时器隔一段时间就运行一次智能体。做得马虎的话，这就是失控的布谷鸟：每次心跳都叫醒模型，每次运行都要花 token，每次运行都给你发一句“一切正常”。到第三条消息时你就不再看了，而真正要紧的那条也就没人读了。

## 解决方案

心跳：一个定时器，把一个固定的提示词 `checkPrompt` 交给智能体，就像有人打了这行字一样。让它有用而不是扰人的，是你在这个定时器周围加的东西：

- **叫醒之前先检查**
   localCondition
   在每次心跳时、调用模型之前运行的普通代码：读一个指标、一个文件、一行数据。只要它说“不”，这次心跳就一分钱都不花。
- **一次只跑一个**
   内置
   上一次运行还没结束时触发的心跳会被直接丢弃，而不是排队等待。两次运行永远不会同时共用历史记录。
- **保险丝**
   maxTicks, timeoutMs, runTimeoutMs
   心跳次数、实际时长和单次运行时长各有一个预算，这样即使你忘了停掉它，它也会自己结束。
- **只说一次，然后关掉**
   stopHeartbeat()
   只在真的发生了什么时才给人发消息，并说明结果如何。工作结束后，就把心跳关掉。

第一条承担了大部分工作。大多数心跳都发现没什么可做，而判断这一点用不着模型：一个 `if` 就够了。在动画里，五次心跳过去，只有一次调用了模型。

## 角色

还是那群熟悉的角色，这次住进了一只口袋宠物里。

- **时钟** (心跳): 它的闹铃每两小时响一次，心跳开着时就一直亮着。
- **括号** (localCondition): 每次心跳都在那些心周围闪一下：那是你自己的代码在读指标。不涉及任何模型。
- **Astor** (循环): 在小垫子上打盹，直到某次心跳说有项指标偏低，然后像往常一样跑循环。
- **神谕者** (模型): 一直睡着，直到 Astor 把 checkPrompt 带给它。
- **图标** (工具): 状态、食物、游戏和呼叫灯：`check_status`、`feed`、`play` 和 `beep_owner`。
- **主人** (人): 一整天都在学校。只收到一声哔，而且已经是好消息了。

在 EventBus 面板里，安静的心跳只有你的代码在运行：对于被你的检查否决的心跳，astorlm 不会发出任何事件。`heartbeat_tick` 只出现一次，就是模型真正被叫醒的时候。

## 代码

**使用 astorlm：**给智能体传入 `heartbeat`，它就会自己启动。各种防护都是选项：`localCondition`、`maxTicks` 和 `timeoutMs`；重叠的心跳会被自动丢弃。主人回来时，你的应用调用 `stopHeartbeat()`。

**从零手写：**在第 2 关的循环外面套一个定时器。用一个标志位防止运行重叠，用一个计数器和一个截止时间当保险丝，再用一个普通函数决定到底要不要调用模型。

**使用 astorlm**

```ts
import { OpenAIProvider, createLocalAgent, tool } from 'astorlm'
import { z } from 'zod'

const HOUR = 60 * 60_000

const checkStatus = tool({
  name: 'check_status',
  description: 'Open the status screen: hunger and happiness in hearts, and whether the pet is sick or asleep.',
  schema: z.object({}),
  execute: async () => pet.status(), // your code: the pet lives in your app
})

const feed = tool({
  name: 'feed',
  description: 'Feed the pet a meal or a snack. A meal fills hunger; a snack only cheers it up.',
  schema: z.object({ food: z.enum(['meal', 'snack']) }),
  execute: async ({ food }) => pet.feed(food),
})

const play = tool({
  name: 'play',
  description: 'Play the left-or-right game with the pet. Winning fills happiness.',
  schema: z.object({}),
  execute: async () => pet.play(),
})

const beepOwner = tool({
  name: 'beep_owner',
  description: 'Beep the owner with a short message. They are at school: only when something happened.',
  schema: z.object({ text: z.string() }),
  execute: async ({ text }) => {
    await sendPush(text) // your code
    return 'Beeped.'
  },
})

const agent = await createLocalAgent({
  // Any OpenAI-compatible endpoint: OpenAI, Ollama, LM Studio, vLLM, a proxy…
  provider: new OpenAIProvider({
    baseURL: 'http://localhost:11434/v1', // e.g. Ollama's default address
    model: 'your-model', // e.g. 'llama3.1', 'gpt-4o-mini'
    apiKey: 'YOUR_API_KEY', // local servers usually ignore it
  }),
  tools: [checkStatus, feed, play, beepOwner],
  maxTurns: 8,
  // Starts on its own as soon as the agent is created. Nobody types anything.
  heartbeat: {
    intervalMs: 2 * HOUR,
    checkPrompt: 'Check on Milonga. Take care of whatever is low, then beep her owner with one line.',
    // Runs on every tick, before the model. While it says no, a tick costs 0 tokens.
    localCondition: () => pet.hunger <= 1 || pet.happy <= 1,
    maxTicks: 6, // the fuses: a school day of ticks at most…
    timeoutMs: 10 * HOUR, // …and of wall-clock time
  },
})

// The owner is home: the app takes over and switches the heartbeat off.
onOwnerHome(() => agent.stopHeartbeat())

// Quiet ticks emit nothing. The ones that wake the model do:
agent.on('event', (event) => {
  if (event.type === 'heartbeat_tick') console.log('heartbeat woke the agent')
})
```

**TypeScript**

```ts
// A heartbeat, from scratch. Plain fetch and timers, no SDK.

// Any OpenAI-compatible endpoint: OpenAI, Ollama, LM Studio, vLLM, a proxy…
const LLM = {
  baseURL: 'http://localhost:11434/v1', // e.g. Ollama's default address
  model: 'your-model', // e.g. 'llama3.1', 'gpt-4o-mini'
  apiKey: 'YOUR_API_KEY', // local servers usually ignore it
}

const HOUR = 60 * 60_000

// 1. The tools, as in level 3. The pet lives in your app.
type ToolFn = (args: Record<string, string | number>) => Promise<string>
const tools: Record<string, ToolFn> = {
  check_status: async () => pet.status(),
  feed: async ({ food }) => pet.feed(String(food)),
  play: async () => pet.play(),
  beep_owner: async ({ text }) => {
    await sendPush(String(text)) // your code
    return 'Beeped.'
  },
}
const toolSchemas = [/* one JSON Schema per tool: check_status(), feed(food), play(), beep_owner(text) */]

type ToolCall = { id: string; function: { name: string; arguments: string } }
type Message =
  | { role: 'user'; content: string }
  | { role: 'assistant'; content: string | null; tool_calls?: ToolCall[] }
  | { role: 'tool'; tool_call_id: string; content: string }

// 2. The loop from level 2, unchanged. Each tick that gets through is one run of it.
async function runAgent(prompt: string, maxTurns = 8): Promise<string> {
  const messages: Message[] = [{ role: 'user', content: prompt }]
  for (let turn = 1; turn <= maxTurns; turn++) {
    const res = await fetch(`${LLM.baseURL}/chat/completions`, {
      method: 'POST',
      headers: { 'content-type': 'application/json', authorization: `Bearer ${LLM.apiKey}` },
      body: JSON.stringify({ model: LLM.model, messages, tools: toolSchemas }),
    })
    const [choice] = (await res.json()).choices
    const reply: Message = choice.message
    messages.push(reply)
    if (choice.finish_reason !== 'tool_calls') return reply.content ?? ''

    for (const call of reply.tool_calls ?? []) {
      const run = tools[call.function.name]
      let output = `Unknown tool: ${call.function.name}`
      try {
        if (run) output = await run(JSON.parse(call.function.arguments))
      } catch (err) {
        output = `Error: ${err instanceof Error ? err.message : err}`
      }
      messages.push({ role: 'tool', tool_call_id: call.id, content: output })
    }
  }
  throw new Error(`No answer after ${maxTurns} turns`)
}

// 3. The heartbeat: a timer, a cheap check before the model, and fuses.
const CHECK_PROMPT = 'Check on Milonga. Take care of whatever is low, then beep her owner with one line.'
const MAX_TICKS = 6

// Plain code, no model: while it says no, a tick costs 0 tokens.
const localCondition = (): boolean => pet.hunger <= 1 || pet.happy <= 1

let running = false
let ticks = 0

async function tick(): Promise<void> {
  if (running) return // still busy with the last tick: skip this one, never overlap
  if (++ticks > MAX_TICKS) return stop() // fuse: a budget of ticks
  if (!localCondition()) return // nothing low: let the model sleep

  running = true
  try {
    console.log(await runAgent(CHECK_PROMPT))
  } catch (err) {
    console.error('heartbeat run failed:', err) // log it, and let the next tick try again
  } finally {
    running = false
  }
}

const timer = setInterval(tick, 2 * HOUR)
const deadline = setTimeout(stop, 10 * HOUR) // fuse: wall-clock time

function stop(): void {
  clearInterval(timer)
  clearTimeout(deadline)
}

// The owner is home: the app takes over.
onOwnerHome(stop)
```

**Python**

```python
# A heartbeat, from scratch. Standard library only, no SDK.
import json
import threading
import time
import urllib.request

# Any OpenAI-compatible endpoint: OpenAI, Ollama, LM Studio, vLLM, a proxy...
LLM = {
    "base_url": "http://localhost:11434/v1",  # e.g. Ollama's default address
    "model": "your-model",  # e.g. "llama3.1", "gpt-4o-mini"
    "api_key": "YOUR_API_KEY",  # local servers usually ignore it
}

HOUR = 60 * 60

def post(path, payload):
    request = urllib.request.Request(
        f"{LLM['base_url']}{path}",
        data=json.dumps(payload).encode(),
        headers={"Content-Type": "application/json", "Authorization": f"Bearer {LLM['api_key']}"},
    )
    with urllib.request.urlopen(request) as response:
        return json.load(response)

# 1. The tools, as in level 3. The pet lives in your app.
def beep_owner(text):
    send_push(text)  # your code
    return "Beeped."

TOOLS = {
    "check_status": lambda: pet.status(),
    "feed": lambda food: pet.feed(food),
    "play": lambda: pet.play(),
    "beep_owner": beep_owner,
}
TOOL_SCHEMAS = [...]  # one JSON Schema per tool: check_status(), feed(food), play(), beep_owner(text)

# 2. The loop from level 2, unchanged. Each tick that gets through is one run of it.
def run_agent(prompt, max_turns=8):
    messages = [{"role": "user", "content": prompt}]

    for _ in range(max_turns):
        choice = post("/chat/completions", {"model": LLM["model"], "messages": messages, "tools": TOOL_SCHEMAS})["choices"][0]
        reply = choice["message"]
        messages.append(reply)
        if choice["finish_reason"] != "tool_calls":
            return reply.get("content") or ""

        for call in reply.get("tool_calls", []):
            try:
                output = TOOLS[call["function"]["name"]](**json.loads(call["function"]["arguments"]))
            except Exception as err:
                output = f"Error: {err}"
            messages.append({"role": "tool", "tool_call_id": call["id"], "content": output})

    raise RuntimeError(f"No answer after {max_turns} turns")

# 3. The heartbeat: a timer, a cheap check before the model, and fuses.
CHECK_PROMPT = "Check on Milonga. Take care of whatever is low, then beep her owner with one line."
INTERVAL = 2 * HOUR
MAX_TICKS = 6
DEADLINE = time.monotonic() + 10 * HOUR  # fuse: wall-clock time
stopped = threading.Event()
on_owner_home(stopped.set)  # the owner is home: the app takes over

def local_condition():
    """Plain code, no model: while it says no, a tick costs 0 tokens."""
    return pet.hunger <= 1 or pet.happy <= 1

# One thread, one run at a time: a tick can never overlap the last one.
ticks = 0
while not stopped.wait(INTERVAL):
    ticks += 1
    if ticks > MAX_TICKS or time.monotonic() > DEADLINE:
        break  # the fuses
    if not local_condition():
        continue  # nothing low: let the model sleep
    try:
        print(run_agent(CHECK_PROMPT))
    except Exception as err:
        print("heartbeat run failed:", err)  # log it, and let the next tick try again
```

## 注意事项

- **astorlm 的心跳只保留一个会话。**每次叫醒模型的心跳都会往同一段历史记录里添东西，所以频繁叫醒模型的心跳，会让它的上下文和账单随每次运行一起增长。让 checkPrompt 保持简短、加上压缩，或者每次运行都换一个全新的智能体（上面的从零手写版本就是这么做的）。
- **留意单次运行的超时。**`runTimeoutMs` 默认是 60 秒。如果某次运行的工具要等很慢的东西，就需要更长的时间，否则会在半路被中止。
- **心跳不是 cron 任务。**它活在你的进程里：进程一停，心跳也停，错过的心跳不会补回来。对于必须挺过重启的工作，让一个真正的调度器来启动智能体，并保留同样的防护。
- **写提示词之前，先决定什么值得发消息。**要写“如果你不得不做了什么，就通知我”，而不是“告诉我进展如何”。一条什么都没说的消息，只会让人学会忽略下一条。
- **独自行动同样需要边界。**没人在看着。喂宠物没问题；任何撤不回来的事，都应该等一个人点头。

## 相关模式

- [4 · 何时停止](https://harnesspatterns.dev/zh/patterns/when-to-stop.md)
- [7 · 背包装满了](https://harnesspatterns.dev/zh/patterns/compaction.md)
- [10 · 每圈重新开始](https://harnesspatterns.dev/zh/patterns/fresh-laps.md)
- [13 · 人在回路](https://harnesspatterns.dev/zh/patterns/human-in-the-loop.md)
- [11 · 可观测性与评估](https://harnesspatterns.dev/zh/patterns/observability.md)