从零到一手撸 Agent 系列 - 第 3 篇：Agent 的心跳 - 主循环与 System Prompt

从零到一手撸 Agent 系列 — 第 3 篇：Agent 的"心跳" — 主循环与 System Prompt

开篇：前两篇搭好了骨架和大脑，现在是让它们一起工作的时候

回想一下进度：

• 第一篇：我们搭了 CLI 骨架（Cobra 三子命令 + .env 加载）
• 第二篇：我们实现了 Provider 层（OpenAI / Anthropic 双后端 + 流式传输）

现在，这两个模块是割裂的。我们需要一个"指挥官"把它们串起来——Agent 主循环。这是整个项目最核心的代码：它决定何时调用 LLM、何时执行工具、何时停下来说"我做完了"。

本篇写完，你的 Agent 将真正拥有"心跳"。

一、Agent 结构体：这个"指挥官"手里有什么？

type Agent struct {cfgConfig// 配置（模型、token 上限等）prov provider.Provider // LLM 后端registry *tools.Registry // 工具注册表checkerpermission.Checker// 权限检查器hooksToolHooks // 事件钩子messages []provider.Message// 对话历史（核心状态）sink event.Sink// 事件输出（终端/UI）// ... 后面文章会讲的：压缩状态、记忆、session 等}

关键在 messages：它是 Agent 的唯一状态。整个对话的上下文都在这一个切片里——system prompt、用户输入、LLM 回复、工具调用、工具结果——全部是 []provider.Message。Agent 的每一次"思考"，就是在这个切片末尾追加新消息，然后整个丢给 LLM。

构造 Agent 使用 Option 模式（Go 惯用的依赖注入）：

a, err := agent.NewAgent(cfg,agent.WithRegistry(registry),// 注入工具agent.WithChecker(checker),// 注入权限agent.WithHooks(hookRunner), // 注入钩子agent.WithSink(sink),// 注入事件输出)

每个 WithXxx 返回一个 Option 接口的实现，内部只是把值赋给 Agent 结构体。好处是：新增依赖不需要修改 NewAgent 的函数签名，调用方也只传自己关心的选项。

二、Run()：外循环 — 一切从这里开始

func (a *Agent) Run(ctx context.Context, userInput string) (string, error) {// 1. 用户输入追加为 user 消息a.messages = append(a.messages, provider.Message{Role:provider.RoleUser,Content: userInput,})// 2. 最多跑 MaxTurns 轮（默认 100）for turn := 0; turn < a.cfg.MaxTurns; turn++ {final, err := a.loopStep(ctx)if err != nil {return "", err}// 3. 如果 loopStep 返回了非空 final，说明 LLM 给出了最终答案if final != "" {a.SaveCurrentSession()// 自动保存return final, nil}}// 4. 超过 MaxTurns：强制结束return "", fmt.Errorf("%w (limit=%d)", ErrMaxTurnsExceeded, a.cfg.MaxTurns)}

逻辑非常清晰：

用户输入 → user 消息追加到历史↓for 循环（最多 MaxTurns 次）：loopStep() → 返回 ""（还有下一轮）or "最终回答"（结束）↓最终回答 or 超限报错

MaxTurns 的存在是安全阀——防止 LLM 陷入"调工具→没结果→再调→又没结果"的死循环。默认 100 轮，对绝大多数场景足够了；如果你的任务是"帮我重构整个项目"，LLM 可能会跑很多轮，可以 --max-turns 200 调大。

三、loopStep()：一轮"心跳"

这是整个项目最重要的函数，我们来逐段拆解。

3.1 发起 LLM 请求

func (a *Agent) loopStep(ctx context.Context) (final string, err error) {// onText: 每收到 LLM 流式输出的一个字就推送给 UIonText := func(s string) {a.sink.Emit(event.Event{Kind: event.Text, Text: s})}// 修复孤儿 tool 消息，防止 API 400a.messages = provider.SanitizeToolPairing(a.messages)// 检查是否需要压缩上下文a.maybeCompact(ctx, a.lastPromptTokens)// 构造请求req := a.buildRequest()// 调用 LLM（流式）ch, err := a.prov.Stream(ctx, req)if err != nil {// 上下文溢出？尝试紧急压缩后重试if provider.IsPromptTooLong(err) && !a.hasAttemptedReactiveCompact {a.reactiveCompact(ctx)a.hasAttemptedReactiveCompact = truereturn "", nil// 回到 Run 的 for 循环重试}return "", fmt.Errorf("调用 LLM 失败: %w", err)}// ...}

3.2 收集流式结果

msg, usage, streamErr := provider.CollectWithText(ch, onText)if streamErr != nil {// 流中断但已有部分输出？注入恢复 prompt 重试if provider.IsStreamInterrupted(streamErr) {if msg.Content != "" {a.messages = append(a.messages, provider.Message{Role: provider.RoleAssistant, Content: msg.Content,})}a.messages = append(a.messages, provider.Message{Role: provider.RoleUser,Content: "[流式传输中断] 请从断点继续，不要重复已输出的内容。",})return "", nil// 回到 for 循环重试}return "", fmt.Errorf("调用 LLM 失败: %w", streamErr)}

流中断恢复的策略值得注意：不是直接重试（那样 LLM 会重新输出已经显示给用户的内容），而是把已输出的部分当作 assistant 消息存入历史，然后追加一条"请从断点继续"的 user 消息——让 LLM 无缝续写。

3.3 判断：文本回答还是工具调用？

a.messages = append(a.messages, msg)// 情况 A：纯文本 → LLM 已经给出最终答案if len(msg.ToolCalls) == 0 {// 终答守卫：检查是否有未完成的 todoif force := a.checkTodoGuard(ctx); force != "" {a.messages = append(a.messages, provider.Message{Role: provider.RoleUser, Content: force,})return "", nil// 继续循环，让 LLM 完成任务}// Stop hook：允许外部脚本在 Agent 结束时干预if a.hooks != nil {if force, ok := a.hooks.Stop(ctx, a.messages); ok {a.messages = append(a.messages, provider.Message{Role: provider.RoleUser, Content: force,})return "", nil}}return msg.Content, nil// ← 最终答案！}// 情况 B：有 tool_calls → 执行工具，然后下一轮继续a.executeBatch(ctx, msg.ToolCalls)return "", nil// ← 返回空，告诉 Run() 继续循环}

这就是 Agent 的核心判断：LLM 回复里有没有 tool_calls？

• 没有 → LLM 认为任务已完成，返回纯文本答案
• 有 → LLM 希望调用工具获取更多信息，执行完工具后回到循环开头，带着工具结果再次请求 LLM

3.4 完整流程图

loopStep():┌─→ 修复孤儿 tool 消息│ 检查是否需要压缩│ 构造请求 → Stream()│ ││ ├─ 错误？→ IsPromptTooLong？→ 紧急压缩 → 重试│ │ 其他错误 → 报错退出│ ││ └─ 成功 → CollectWithText()│ ││ ├─ StreamInterrupted？→ 保存已输出内容 → 注入恢复提示 → 重试│ ││ └─ 正常 → 追加 assistant 消息到 history│ ││ ├─ 无 tool_calls？→ 终答守卫 → Stop hook → 返回 final！│ ││ └─ 有 tool_calls？→ executeBatch() → 返回空（继续循环）│↓└────────────────────────────────────────────────┘

四、invokeTool：执行一个工具

func (a *Agent) invokeTool(ctx context.Context, tc provider.ToolCall) string {name := tc.Name// 1. 解析 JSON 参数var args map[string]anyjson.Unmarshal([]byte(tc.Arguments), &args)// 2. PreToolUse hook：外部脚本可在工具执行前拦截if a.hooks != nil {if blocked, msg := a.hooks.PreToolUse(ctx, name, args); blocked {return "Blocked by hook: " + msg}}// 3. 权限检查：DenyList → BashAsk → WorkdirBoundary 三级管线if a.checker != nil {r := a.checker.Check(ctx, name, args)if r.Decision == permission.DecisionDeny {return "Permission denied: " + r.Reason}}// 4. 查找并执行工具tool := a.registry.Get(name)if tool == nil {return fmt.Sprintf("Error: 工具 %q 未注册", name)}out, err := tool.Execute(ctx, args)// 5. PostToolUse hookif a.hooks != nil {a.hooks.PostToolUse(ctx, name, args, out)}return out// 成功返回工具输出，失败返回 "Error: ..."}

调用链路：LLM 提议 → PreToolUse Hook → 权限检查 → 工具执行 → PostToolUse Hook → 结果回传。每个环节都可能阻断，但通过把结果统一格式化为字符串，上层完全不用区分是成功、被阻断还是错误——全部作为 tool result 喂回 LLM。

五、executeBatch：批量执行 + 并行优化

当 LLM 一次性请求多个工具时（比如同时读 3 个文件），Agent 会智能地划分并行批次：

func (a *Agent) executeBatch(ctx context.Context, calls []provider.ToolCall) {results := make([]string, len(calls))// 将 tool_calls 按是否只读划分为串行/并行批次for _, batch := range partitionToolCalls(a.registry, calls) {if batch.parallel {runParallel(batch.start, batch.end, results, func(i int) {results[i] = a.invokeTool(ctx, calls[i])})} else {// 串行执行for i := batch.start; i < batch.end; i++ {results[i] = a.invokeTool(ctx, calls[i])}}}// 将所有结果作为 tool 消息追加到历史for i, tc := range calls {a.messages = append(a.messages, provider.Message{Role: provider.RoleTool,ToolCallID: tc.ID,Name: tc.Name,Content:results[i],})}}

分批次的关键：

func partitionToolCalls(r *tools.Registry, calls []provider.ToolCall) []toolCallBatch {var batches []toolCallBatchfor i := 0; i < len(calls); {if isParallelisable(r, calls[i].Name) {// 连续的只读工具作为一个并行批次start := ifor i < len(calls) && isParallelisable(r, calls[i].Name) { i++ }batches = append(batches, toolCallBatch{start, i, parallel: true})} else {// 写操作必须串行，单独一个批次batches = append(batches, toolCallBatch{start: i, end: i + 1})i++}}return batches}

假设 LLM 说："同时读 A.go、B.go、C.go，然后修改 D.go"：

calls = [read(A), read(B), read(C), write(D)]↓ partitionToolCallsbatches = [{read(A), read(B), read(C)}← 并行（3 goroutine 同时读）{write(D)} ← 串行（等上面跑完再写）]

写工具不能并行——文件可能互相冲突。maxParallelTools = 8 限制并发数，防止把系统资源打满。

六、System Prompt：告诉 LLM 它是什么、能做什么

LLM 需要一个"人设"——这就是 System Prompt。Agent 根据注册的工具列表自动生成：

func buildSystemPrompt(registry *tools.Registry, skills []skill.Skill) string {var b strings.Builderb.WriteString("你是一个编码助手，可以使用以下工具完成任务。nn")toolList := registry.List()for i, t := range toolList {fmt.Fprintf(&b, "%d. %sn", i+1, t.Name())fmt.Fprintf(&b, " 描述: %sn", t.Description())fmt.Fprintf(&b, " 参数 schema: %sn", string(t.Schema()))b.WriteString("n")}b.WriteString("请按用户意图选择合适的工具，按需连续调用多个工具以完成任务。n")// 有特定工具时追加专项指导if hasTodoTools(registry) {b.WriteString(todoGuidance) // "用 todo_write 管理多步骤任务..."}if hasTaskTool(registry) {b.WriteString(taskGuidance) // "用 task 委派子任务给子 agent..."}return b.String()}

这意味着：你每注册一个新工具，System Prompt 自动更新。LLM 立刻就知道多了什么能力。

还有一些专项指导会在特定工具存在时自动追加——比如注册了 todo_write 和 complete_step 后，System Prompt 里就会多一段关于如何用这两个工具管理多步骤任务的详细说明。

七、事件系统（Event Sink）：Agent 如何与外界对话

Agent 不直接写 fmt.Print——所有输出都通过 event.Sink 接口发出：

type Sink interface {Emit(Event)}type Kind intconst (Text Kind = iota// LLM 输出的文本增量ToolDispatch// 开始执行工具ToolResult// 工具执行结果ApprovalRequest // 请求用户审批TurnDone// 一轮对话结束Notice// 系统通知)

不同的"外壳"实现不同的 Sink：

• REPL 模式：TextSink{Out: os.Stdout} — 直接写到终端
• TUI 模式：TuiSink — 通过 channel 推送到 Bubble Tea UI

这种"依赖倒置"让 Agent 核心完全不感知 UI 形态——你可以给它套 REPL，也可以套 TUI，甚至可以套 WebSocket 做远程 Agent。

八、终答守卫（Todo Guard）：防止 Agent "假装做完"

一个常见问题：Agent 说"完成了！"，但其实 todo_write 里还有好几项 in_progress。

终答守卫在 Agent 准备输出最终答案之前检查：

func (a *Agent) checkTodoGuard(ctx context.Context) string {todos := ledger.CurrentTodos()var incomplete []stringfor _, t := range todos {if t.Status != "completed" {incomplete = append(incomplete, t.Content+" ["+t.Status+"]")}}if len(incomplete) == 0 { return "" }// 全完成，放行// 有未完成任务 → 拒绝这次"最终回答"，要求 LLM 继续工作return fmt.Sprintf("宿主就绪检查失败: %d/%d 待办仍未完成。请先完成剩余任务...",len(incomplete), len(todos))}

如果 LLM 反复"假装做完"，守卫最多阻断 maxGuardBlocks=3 次，然后放行——避免死循环。

九、完整串联：现在让 Agent 跑起来

把第一篇的 once 命令真正接上 Agent：

func runOnce(cmd *cobra.Command, args []string) error {workdir, _ := os.Getwd()cfg := buildConfig(cmd)// 1. 构造工具注册表registry := tools.DefaultRegistry(workdir)// DefaultRegistry 里预装了 read_file、write_file、edit_file、// glob_file、grep、bash、web_fetch 等基础工具// 2. 构造 Agenta, err := agent.NewAgent(cfg,agent.WithRegistry(registry),agent.WithSink(&event.TextSink{Out: os.Stdout, Err: os.Stderr}),)if err != nil { return err }// 3. 运行！final, err := a.Run(cmd.Context(), onceMessage)if err != nil { return err }fmt.Println("n--- 最终回答 ---")fmt.Println(final)return nil}

编译运行：

go build -o coding-agent ./cmd./coding-agent once -m "列出当前目录所有 .go 文件"

你会看到：

[coding-agent] running once...→ Agent 调用 glob_file(pattern="*.go")→ glob_file 返回文件列表→ Agent 输出："当前目录有 3 个 Go 文件：main.go、root.go、once.go"

你的 Agent 第一次"主动干活"了！ 它不是直接回答，而是意识到需要先调用工具，拿到结果后再组织回答。这才是 Coding Agent 的本质。

小结

这篇我们实现了整个项目的心跳机制——Agent 主循环。

下篇预告：我们将深入工具系统——Tool 接口的契约设计、Registry 注册表模式、以及前三个工具（read_file、write_file、glob_file）的完整实现。这是让 Agent 真正"有手可用"的一篇。