从零构建一个 AI Agent:分享我做 PulseBot 的全过程

在动手写代码之前，我想先聊聊 Agent 的基本构成，因为很多初学者会把 Agent 和”聊天机器人”搞混。

一个 Agent，不管框架怎么包装，本质上就是四个东西：

LLM（大语言模型）——Agent 的”大脑”。它负责理解用户的意图、做出决策、生成回复。PulseBot 支持接入 Anthropic Claude、OpenAI、Ollama、OpenRouter 和 NVIDIA 的模型，选择哪个取决于你的场景和预算。

Agent Loop（智能体循环）——这是 Agent 区别于普通聊天机器人的关键。聊天机器人是一问一答，问完就结束了。Agent Loop 是一个持续运转的循环：接收输入 → 思考 → 调用工具 → 观察结果 → 继续思考 → 输出回复。这个循环可以跑多轮，中间可能会调用好几个工具，直到 Agent 觉得任务完成了才停下来。

Memory（记忆）——这又分两层。短期记忆就是当前对话的上下文，比如你说了什么、Agent 回了什么，这些信息会一直保留在对话窗口里。长期记忆则是跨会话的——你上周告诉 Agent 你用的是 PostgreSQL，下次再聊数据库的时候，它应该还记得这件事。没有长期记忆的 Agent，每次对话都像第一次见面，用起来会很痛苦。

Tool（工具）——Agent 的”手脚”。LLM 本身只能生成文字，但通过 Tool Calling，它可以读写文件、运行命令、查询数据库、发送消息、调用 API。工具的丰富程度直接决定了 Agent 能做多少事情。

这四个部分，每一个都有很多实现方式。下面我会用 PulseBot 的实际设计来说明，我是怎么把它们组合到一起的。

这里简单解释一下 Timeplus 的"流"（Stream）是什么，因为它和传统数据库里的"表"有本质区别。

在传统数据库中，表是静态的——数据写进去之后就安静地躺在那里，你执行一条 SELECT，它返回结果，查询就结束了。Timeplus 的流不是这样。流是一个持续增长的、无界的事件序列。你对一个流执行 SELECT，查询不会结束——它会一直运行着，每当有新事件写入这条流，查询就立刻输出新的结果。这种行为在 Timeplus 里叫做"流式查询"（Streaming Query），它回答的不是"过去发生了什么"，而是"现在正在发生什么"。

当然，如果你确实需要查历史数据，Timeplus 也支持。只要用 table() 函数把流包一层——比如 SELECT * FROM table(messages)——查询就变成了传统的有界查询，返回流中已有的全部历史数据，查完即止。

所以 Timeplus 的流同时扮演了两个角色：它既是一个实时的消息通道（新事件到达时立刻可查），又是一个持久化的存储（历史事件随时可回溯）。这也是为什么 PulseBot 能用流来同时解决消息传递、状态存储和实时监控这三个看似不同的问题——因为在 Timeplus 的世界里，它们本来就是同一个东西。

整个框架维护了 9 条 Timeplus 流，各有分工：

• messages 是所有对话的中心枢纽，用户输入、Agent 回复、工具调用的结果都在这里。

• llm_logs 记录每次 LLM 调用的细节——用了哪个模型、花了多少 token、延迟多久、第一个 token 什么时候出来、调用是否成功。

• tool_logs 记录每次工具执行的参数、耗时和结果。

• memory 存放向量化的长期记忆，带有重要性评分。

• events 是系统的”神经系统”，承载心跳、通道连接、技能加载、告警、任务通知等信号

• task_triggers 是定时任务的审计日志。

• 还有三条 kanban 相关的流（kanban、kanban_projects、kanban_agents），用于多 Agent 协作。

第一步，监听。 Agent 持续监听 messages 流上的 user_input 事件。注意，这里不是轮询（polling），而是流驱动的——有新消息到达时才触发处理，没消息的时候 Agent 不消耗计算资源。

第二步，构建上下文。 Agent 从 messages 流中拉取最近的对话历史（短期记忆），同时通过向量搜索从 memory 流中检索最相关的长期记忆。这两部分合在一起，构成了这次 LLM 调用的上下文。

第三步，调用 LLM。 把构建好的上下文连同所有可用的工具定义一起发给大模型，让它做推理。

第四步，执行工具。 如果 LLM 的回复里包含了工具调用请求（tool call），Agent 就执行对应的工具，并把每次执行的参数、耗时和结果写入 tool_logs 流。工具执行完之后，结果会回传给 LLM，让它决定是否需要继续调用其他工具，还是直接生成最终回复。

第五步，写回结果。 把 Agent 的回复写入 messages 流，同时把这次 LLM 调用的完整信息（模型名称、提供商、输入/输出 token 数、预估费用、延迟、首 token 时间、调用的工具列表、成功/失败状态）写入 llm_logs 流。

第六步，提取记忆。 LLM 会在回复过程中自动识别对话中值得记住的信息——用户的技术偏好、项目背景、做出的决定——然后附带一个重要性评分，写入 memory 流。

这个六步循环是异步执行的（基于 Python 的 async/await），整个过程完全可以通过 SQL 查询来观察。比如你想知道过去一小时 Agent 花了多少钱，直接查 llm_logs 就行；想知道哪个工具调用最慢，查 tool_logs 排个序就够了。不需要额外接 Prometheus 或者 Grafana，流数据库本身就是你的监控系统。

在提示词设计上，系统提示词的结构大致是这样的：先描述 Agent 的身份和行为规范；然后列出所有已发现技能的名称与描述（总共可能只有几百个 token）；接着给出可用工具的定义（JSON Schema 格式）；最后注入通过语义搜索检索到的相关长期记忆。这四个部分组合起来，构成了每次对话轮次中动态生成的完整提示词。

另外，系统提示词里只放技能的名字和简短描述，而不是完整的指令。每个技能在提示词中大约只占 24 个 token。当用户的问题匹配到某个技能时，Agent 会动态加载那个技能的完整指令——这是一种”懒加载”模式，既保持了基础上下文的精简，又不牺牲深度能力。

ulseBot 的技能系统遵循 http://agentskills.io 标准，并兼容 OpenClaw/ClawHub 的技能格式。一个技能的结构非常简单——就是一个文件夹，里面有一个 SKILL.md 文件。这个文件的头部是 YAML frontmatter，定义了名称、描述、版本和依赖关系，正文部分是给 Agent 看的 Markdown 指令。

Agent Skill的设计让写技能变得像写文档一样简单。你不需要学新的 DSL，不需要注册什么 API 端点，只要把一个 Markdown 文件丢进指定目录，PulseBot就能发现它，并且无需重启，实现了热加载。

PulseBot 自带了五个内置技能：

file_ops 提供文件操作——read_file、write_file、list_directory， Agent 可以在受限的目录范围内读写文件。

shell 提供命令行执行能力，但加了安全防护，会拒绝执行危险操作（比如 rm -rf /）。

workspace 用于创建和发布交互式 Web 应用（后面会专门讲）。

scheduler 负责定时/周期任务的管理。

project_manager 处理多 Agent 项目的协调。

每个技能可以声明自己提供的工具（Tool），这些工具通过标准的 function calling 机制暴露给 LLM。当 LLM 决定调用某个工具时，Agent 框架会负责参数验证、执行调用、日志记录和结果回传。

除了本地技能，PulseBot 还集成了 ClawHub 注册中心。你可以用 CLI 命令搜索和安装社区发布的技能：

pulsebot skill install timeplus-sql-guide

以 timeplus-sql-guide 这个技能为例，它包含了 Proton 流处理 SQL 的完整参考——tumble()、hop()、session() 窗口函数的用法，外部流（Kafka/Redpanda/Pulsar）的创建语法，Python/JavaScript UDF 的编写方式，RANDOM STREAM 测试数据生成，以及 CREATE TASK 定时任务语法。这些内容加起来可能有好几千个 token，但在用户不问相关问题的时候，它们完全不占用上下文空间。只有当用户说”帮我写一个流式聚合查询”的时候，Agent 才会去加载这个技能的完整指令。

workspace: base_dir: "./workspaces" port: 8001 api_key: "your-internal-key"

定时任务是很多 Agent 场景的刚需——每天早上 9 点总结一下昨天的 GitHub 活动，每小时检查一次服务器状态，每周生成一份数据报告。

大多数 Agent 框架处理定时任务的方式是用系统 cron 或者在进程内跑一个定时器。这两种方案都有明显的问题：系统 cron 和 Agent 框架是脱节的，你很难在框架层面管理和监控这些任务；进程内定时器则在 Agent 重启后就丢失了。

PulseBot 选择用 Timeplus 原生的 TASK对象来做定时任务。

具体机制是这样的。scheduler 技能通过 CREATE TASK SQL 语句在 Timeplus 中创建一个 TASK 对象，这个 TASK 按照指定的间隔或 cron 表达式定期执行。每次执行时，TASK 会调用一个 Python UDF（用户自定义函数），这个 UDF 向 PulseBot API 的 /api/v1/task-trigger 端点发送一个 HTTP POST 请求。API 服务器收到请求后，Agent 就会处理关联的 prompt，生成结果，并通过 events 流以 task_notification 事件的形式广播给所有已连接的通道（Telegram、网页聊天等）。每次调用也会记录到 task_triggers 流中，用于审计。

当一个任务复杂到单个 Agent 处理不过来时，就需要多个 Agent 协作。

PulseBot 用了一个大家都很熟悉的比喻来组织多 Agent 协作——看板（Kanban）。一个主 Agent（通常扮演项目经理的角色）可以创建一个”项目”，把任务拆分成多个子任务，分配给不同的 Agent 去执行，然后追踪整体进度。

这套机制依赖三条专用的 Timeplus 流。

kanban 流是 Agent 之间的消息队列。Agent 在这里发布任务项、状态更新和工作成果。每条消息就是一张”卡片”，可以在不同的状态之间流转。

kanban_projects 流记录项目的生命周期——什么时候创建的、当前状态是什么、参与了哪些 Agent、整体目标是什么。

kanban_agents 流维护每个 Agent 的状态和检查点。如果某个 Agent 中途被打断（比如进程重启），它可以从最近的检查点恢复，而不需要从头开始。

项目经理技能提供了四个工具：create_project（创建新项目并分配任务）、list_projects（查看所有进行中的项目）、get_project_status（检查特定项目的进度）、cancel_project（取消一个项目）。

为什么选看板而不是其他协作模式？因为看板天然就是事件驱动的——每一次状态转换都是一条事件。这和 PulseBot 基于流的架构完美契合。每个 Agent 的每一步操作都被记录为流中的事件，这意味着整个协作过程都是可观察、可查询、可回放的。

和基于内存的多 Agent 框架相比，这种设计有一个很实际的好处：Agent 团队可以跨机器分布。主 Agent 和子 Agent 不需要跑在同一台机器上，它们通过 Timeplus 流来通信。只要能连到同一个 Timeplus 实例，就能协作。

这里是Pulsebot用以下提示词创建的一个一次性任务的多Agent团队项目：

Please create a project, prepare an investment memo for this startup ,http://timeplus.com. In parallel: research the market opportunity and analyze the founding team's background. Once both are done, a senior analyst should synthesize them into the core investment thesis. Then an editor polishes the final memo.

区别于n8n,dify等需要用户构建任务图的Agent框架，Pulsebot可以利用自然语言来创建Agent团队的依赖关系非常的方便。

如果你注意到了，前面讲到的多 Agent 协作已经支持两种执行模式——一次性运行（one-shot）和定时/周期运行（scheduled）。但在真实的生产场景中，还有大量需求是这两种模式覆盖不了的。比如：每当 events 流中出现一条 severity = ‘error’ 的事件，就自动触发一组 Agent 去排查问题；或者当某个传感器的读数超过阈值时，立刻启动一个分析流水线。

这类需求的本质是：Agent 工作流的触发源不是人，也不是时钟，而是实时数据流中的某个模式。

为了解决这个问题，我在 PulseBot 中实现了第三种项目执行模式——事件驱动模式（event-driven）。它的触发源是一条用户定义的 Proton 流式 SQL 查询，每当查询返回一行新结果，就触发一次多 Agent 工作流。

EventWatcher：流式查询的”哨兵”

这个机制的核心是一个叫做 EventWatcher 的组件（代码在

pulsebot/agents/event_watcher.py 中）。它的职责很明确：订阅一条用户指定的流式 SQL 查询，从每一行结果中提取指定的上下文字段，判断当前项目是否空闲，如果空闲就触发一次工作流运行。

整个触发流程是这样走的：

1. Proton 流式 SQL 查询（用户定义）持续运行，等待新事件到达

2. 新的一行结果到达，EventWatcher 从中提取 context_field 的值

3. 检查项目是否正在忙碌——如果是，跳过这行，推进 checkpoint，继续等待下一条事件

4. 如果项目空闲，标记为忙碌，写一条 trigger 消息到 kanban 流（包含 msg_type: "trigger"、target_id: "manager_{project_id}" 以及拼接好的提示词），然后推进 checkpoint

5. ManagerAgent（项目经理 Agent）从 kanban 流收到 trigger 消息，把带有上下文的任务分发给各个 Worker Agent

6. Worker Agent 各自执行 LLM 推理和工具调用，把结果通过 kanban 流返回给 ManagerAgent

7. ManagerAgent 汇总所有 Worker 的结果，通过 events 流广播 task_notification，并调用 on_run_complete 回调

8. ProjectManager.mark_project_idle(project_id) 释放忙碌标记，EventWatcher 可以处理下一条事件了

这里有一个关键的设计决策：drop-on-busy（忙碌时丢弃）。如果 Agent 团队正在处理上一条事件触发的工作流，新到达的事件会被直接跳过，而不是排队等待。为什么？因为在实时场景中，堆积未处理的事件往往比丢弃它们更危险——想象一下，一个错误告警在 10 秒内触发了 50 次，你不会希望 Agent 团队把这 50 次告警一条一条地按顺序处理。丢弃重复事件、只处理最新的，通常才是正确的行为。

event_query:    "SELECT payload FROM pulsebot.events WHERE severity = 'error'"context_field:  "payload"trigger_prompt: "检测到系统错误，请调查并总结："

{"source": "agent:xyz", "message": "Connection refused on port 8463"}

EventWatcher 处理（或跳过）每一行之后，都会把当前的 Proton 序列号 _tp_sn 作为 checkpoint 持久化到 kanban_agents 流中（agent_id 为 event_watcher_{project_id}）。

服务重启时会发生什么？ProjectManager 在恢复逻辑（

_recover_scheduled_projects）中会检查每个活跃的 scheduled 项目的 schedule_type。如果是 'interval' 或 'cron'，走原来的 Timeplus Task 恢复路径。如果是 'event'，则从 kanban_agents 读取 EventWatcher 的最后 checkpoint，然后重新构建流式查询，加上 _tp_sn > {checkpoint_sn} 的过滤条件和 SETTINGS seek_to='earliest'，从上次处理到的位置继续。

具体来说，EventWatcher 构建查询的逻辑是：

如果没有 checkpoint（全新启动），查询加上 SETTINGS seek_to='{start_time}'，从项目创建时间开始消费事件。

如果有 checkpoint（重启恢复），在原始查询后面追加 AND _tp_sn > {checkpoint_sn} SETTINGS seek_to='earliest'，从最后处理过的序列号之后开始。

这套机制确保了 exactly-once 语义（至少在 Agent 工作流触发层面）——不会漏掉事件，也不会重复触发。

聊完事件驱动，我想再补充一点关于整体可观测性的内容。

PulseBot 暴露了一个流式 SQL 代理端点（POST /query），接受原始 SQL 查询并返回 NDJSON（换行分隔的 JSON）格式的结果流。这意味着任何外部应用——不管是网页前端还是移动应用——都可以直接对 PulseBot 的任意一条流运行流式查询，实时获取 Agent 的行为数据。

比如你想做一个 Agent 成本监控面板，只需要运行这样一条查询：

SELECT model, sum(input_tokens) as total_input, sum(output_tokens) as total_output,      sum(estimated_cost) as total_costFROM table(llm_logs)GROUP BY model

不需要 Prometheus，不需要 Grafana，不需要额外的数据管道。流数据库本身就是你的可观测性平台。