Joshua's Blog

A programmer's blog

0%

阿里智能体open-code-review源码解析

发表于 更新于 分类于
本文字数： 32k 阅读时长 ≈ 29 分钟

最近AI vibe coding多了发现一些问题，其中包括AI生成代码的review问题。目前我对于AI代码的态度偏谨慎，大部分AI代码会review一遍，但毕竟人工review费时费力，代码量大了之后review就更痛苦了。
于是想到了之前博文提到的让AI去review的方式，尝试了在AI coding ide上通过skill来review，发现效果不太好，最近看到阿里发布了open-code-review的开源智能体，试了下好奇这个agent的逻辑，便写篇文章分析下。

项目地址: open-code-review，分支main，当前版本commit 1c8f930，go语言编写，以下简称ocr，官方介绍：

Open Code Review 是一款 AI 驱动的代码审查 CLI 工具。它的前身是阿里集团内部官方 AI 代码审查助手，过去两年在内部服务了数万开发者，识别了数百万个代码缺陷。经过大规模充分验证后，我们将其孵化为开源项目，对社区开放。只需配置一个模型端点即可使用。
它读取 Git diff，通过具备工具调用能力的 Agent 将变更文件发送至可配置的 LLM，生成具有行级精度的结构化审查意见。Agent 可以读取完整文件内容、搜索代码库、检查其他变更文件以获取上下文，从而进行深度审查——而非仅停留在表面的 diff 反馈。除了 diff 审查，ocr scan 可以审查整个文件，适用于审计不熟悉的代码库或没有有意义 diff 的目录。

源码解析

大致走一遍代码：

main

cmd/opencodereview/main.go

1
2
3
4
5
6
7
8
9
10
11
12
13
14
func main() {
  llm.AppVersion = Version
  llm.InitEmbeddedLoader()

  ctx := context.Background()
  if telemetry.Init(ctx) {
    defer telemetry.ShutdownWithTimeout(ctx, 5*time.Second)
  }

  if err := dispatch(); err != nil {
    fmt.Fprintf(os.Stderr, "Error: %v\n", err)
    os.Exit(1)
  }
}

InitEmbeddedLoader初始化内置分词器，分词器用于估算token量。
telemetry.Init(ctx)初始化opentelemetry，用于后续上报trace数据。

dispatch

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
switch args[0] {
case "--version", "-V":
  printVersion()
  return nil
case "version":
  printVersion()
  return nil
case "review", "r":
  return runReview(args[1:])
case "scan", "s":
  return runScan(args[1:])
case "config":
  return runConfig(args[1:])
case "llm":
  return runLLM(args[1:])
case "rules":
  return runRules(args[1:])
case "viewer":
  return runViewer(args[1:])
case "delegate", "d":
  return runDelegate(args[1:])
case "session", "sessions":
  return runSession(args[1:])
case "-h", "--help":
  printTopLevelUsage()
  return nil
default:
  return fmt.Errorf("unknown command: %s\nRun 'ocr' for usage", args[0])
}

根据命令行参数选择执行命令，这里我们分析runReview

runReview

cmd/opencodereview/review_cmd.go

1
2
3
4
func runReview(args []string) error {
  // ...
  // review path: git repo is required (diff concepts depend on it).
  cc, err := loadCommonContext(opts.repoDir, opts.rulePath, opts.maxTools, opts.maxGitProcs, true)

loadCommonContext准备通用上下文，包含提示词模板、代码目录准备、审查规则、git命令执行器

loadCommonContext

cmd/opencodereview/shared.go

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
func loadCommonContext(repoDirInput, rulePath string, maxTools, maxGitProcs int, requireGit bool) (*commonContext, error) {
  tpl, err := template.LoadDefault()
  // ...

  repoDir, isGit, err := resolveWorkingDir(repoDirInput, requireGit)
  // ...

  resolver, fileFilter, err := rules.NewResolver(repoDir, rulePath)
  // ...

  return &commonContext{
    Template:   tpl,      // 提示词模板
    RepoDir:    repoDir,  // 代码目录
    Resolver:   resolver,  // 审查规则
    FileFilter: fileFilter, // 文件过滤器
    GitRunner:  gitcmd.New(maxGitProcs), // git命令执行器, 控制并发审查时的git命令执行
    IsGitRepo:  isGit,  // 是否是git仓库
  }, nil
}

template.LoadDefault

internal/config/template/template.go

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
func LoadDefault() (*Template, error) {
  data, err := templateFS.ReadFile("task_template.json")

  var m templateManifest
  if err := json.Unmarshal(data, &m); err != nil {
    return nil, fmt.Errorf("unmarshal task_template manifest: %w", err)
  }

  var tpl Template
  tpl.MaxTokens = m.MaxTokens
  tpl.MaxToolRequestTimes = m.MaxToolRequestTimes
  tpl.PlanModeLineThreshold = m.PlanModeLineThreshold

  if tpl.MainTask, err = resolveConversation(m.MainTask); err != nil {
    return nil, fmt.Errorf("MAIN_TASK: %w", err)
  }
  if tpl.PlanTask, err = resolveOptionalConversation(m.PlanTask, "PLAN_TASK"); err != nil {
    return nil, err
  }
  if tpl.MemoryCompressionTask, err = resolveConversation(m.MemoryCompressionTask); err != nil {
    return nil, fmt.Errorf("MEMORY_COMPRESSION_TASK: %w", err)
  }
  if tpl.ReLocationTask, err = resolveOptionalConversation(m.ReLocationTask, "RE_LOCATION_TASK"); err != nil {
    return nil, err
  }
  if tpl.ReviewFilterTask, err = resolveOptionalConversation(m.ReviewFilterTask, "REVIEW_FILTER_TASK"); err != nil {
    return nil, err
  }
  return &tpl, nil
}

加载任务模板，task_template.json在源码目录internal/config/template/task_template.json，内容为json：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
{
  "MAIN_TASK": {
    "timeout": 120,
    "messages": [
      { "role": "system", "prompt_file": "main_task_system.md" },
      { "role": "user", "prompt_file": "main_task_user.md" }
    ]
  },
  "PLAN_TASK": {
    "timeout": 180,
    "messages": [
      { "role": "system", "prompt_file": "plan_task_system.md" },
      { "role": "user", "prompt_file": "plan_task_user.md" }
    ]
  },
  "MEMORY_COMPRESSION_TASK": {
    "timeout": 120,
    "messages": [
      { "role": "system", "prompt_file": "memory_compression_task_system.md" },
      { "role": "user", "prompt_file": "memory_compression_task_user.md" }
    ]
  },
  "REVIEW_FILTER_TASK": {
    "timeout": 180,
    "messages": [
      { "role": "system", "prompt_file": "review_filter_task_system.md" },
      { "role": "user", "prompt_file": "review_filter_task_user.md" }
    ]
  },
  "RE_LOCATION_TASK": {
    "timeout": 180,
    "messages": [
      { "role": "system", "prompt_file": "re_location_task_system.md" },
      { "role": "user", "prompt_file": "re_location_task_user.md" }
    ]
  },
  "MAX_TOOL_REQUEST_TIMES": 30,
  "PLAN_MODE_LINE_THRESHOLD": 50,
  "MAX_TOKENS": 58888
}

template.LoadDefault即是解析上述配置，加载各任务的提示词模板，plan_task是规划用任务，main_task是主review任务，memory_compression_task是上下文压缩任务，review_filter_task是最终review llm输出的任务，prompt文件位于源码目录internal/config/template/prompts下，解析后的go数据结构：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
type LlmConversation struct {
  Timeout  int           `json:"timeout"`
  Messages []ChatMessage `json:"messages"`
}
type Template struct {
  MainTask              LlmConversation  `json:"MAIN_TASK"`
  PlanTask              *LlmConversation `json:"PLAN_TASK,omitempty"`
  MemoryCompressionTask LlmConversation  `json:"MEMORY_COMPRESSION_TASK"`
  MaxTokens             int              `json:"MAX_TOKENS"`
  MaxToolRequestTimes   int              `json:"MAX_TOOL_REQUEST_TIMES"`
  PlanModeLineThreshold int              `json:"PLAN_MODE_LINE_THRESHOLD"`
  ReLocationTask        *LlmConversation `json:"RE_LOCATION_TASK,omitempty"`
  ReviewFilterTask      *LlmConversation `json:"REVIEW_FILTER_TASK,omitempty"`
}

loadCommonContext其他内容不再赘述。

applyCLIExcludes

1
2
3
func runReview(args []string) error {
  // ...
  applyCLIExcludes(cc, splitPaths(opts.excludes))

将命令行参数exclude追加到commonContext的fileFilter中

validateReviewRefs

1
2
3
func runReview(args []string) error {
  // ...
  if err := validateReviewRefs(cc.RepoDir, opts); err != nil {

ocr可以指定review的git commit，这里校验其有效性

getCommitMessage

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
func runReview(args []string) error {
  // ...
  if opts.commit != "" && opts.background == "" {
    if msg, err := getCommitMessage(cc.RepoDir, opts.commit); err == nil && msg != "" {
      opts.background = msg
    }
  }
  if opts.backgroundFile != "" {
    // Resolve relative paths against the git top-level (cc.RepoDir), matching
    // file_read semantics, so `-B ./docs/context.md` works from any directory.
    bgPath := resolveBackgroundFilePath(cc.RepoDir, opts.backgroundFile)
    fileBackground, err := loadBackgroundFile(bgPath)
    if err != nil {
      return err
    }
    opts.background = mergeBackground(opts.background, fileBackground)
  }

background最终会被替换到提示词中，这里将commit的注释和用户指定bgfile读取

runPreview

1
2
3
4
5
func runReview(args []string) error {
  // ...
  if opts.preview {
    return runPreview(cc, opts)
  }

preview模式时，runPreview仅打印要review的文件，不做实际llm交互

loadReviewResumeState

1
2
3
func runReview(args []string) error {
  // ...
  resumeState, err := loadReviewResumeState(cc.RepoDir, opts)

断点续审，读取上次的审查状态

loadLLMRuntime

1
2
3
func runReview(args []string) error {
  // ...
  rt, err := loadLLMRuntime(cc.Template, opts.toolConfigPath, opts.model)

cmd/opencodereview/shared.go

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
func loadLLMRuntime(tpl *template.Template, toolConfigPath, modelOverride string) (*llmRuntime, error) {
  toolEntries, err := toolsconfig.Load(toolConfigPath)
  if err != nil {
    return nil, fmt.Errorf("load tools: %w", err)
  }
  planToolDefs := agent.BuildToolDefs(toolEntries, true)
  mainToolDefs := agent.BuildToolDefs(toolEntries, false)

  appCfg, err := LoadAppConfig(cfgPath)
  var lang string
  if appCfg != nil {
    lang = appCfg.Language
  }
  tpl.ApplyLanguage(lang)
  ep, err := llm.ResolveEndpointWithModelOverride(cfgPath, modelOverride)
  return &llmRuntime{
    Client:       llm.NewLLMClient(ep),
    Model:        ep.Model,
    PlanToolDefs: planToolDefs,
    MainToolDefs: mainToolDefs,
    Collector:    tool.NewCommentCollector(),
    AppCfg:       appCfg,
  }, nil
}

BuildToolDefs读取工具的定义，这里的ToolDefs只有工具的描述信息；LoadAppConfig加载配置；ApplyLanguage会根据语言配置给模板提示词追加一句"\n\nAlways respond in "；ResolveEndpointWithModelOverride解析实际的llm api endpoint。

buildToolRegistry

1
2
3
4
5
6
7
8
9
10
11
func runReview(args []string) error {
  // ...
  mode := tool.ParseReviewMode(opts.from, opts.to, opts.commit)
  ref, _ := mode.RefValue(opts.to, opts.commit)
  fileReader := &tool.FileReader{
    RepoDir: cc.RepoDir,
    Mode:    mode,
    Ref:     ref,
    Runner:  cc.GitRunner,
  }
  tools := buildToolRegistry(rt.Collector, fileReader)
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
const (
  // ModeWorkspace reads files from the current working tree.
  ModeWorkspace ReviewMode = iota
  // ModeRange reads files as they exist at a specific git ref (--to value).
  ModeRange
  // ModeCommit reads files as they exist at a specific commit hash.
  ModeCommit
)
// ParseReviewMode returns the correct ReviewMode based on provided flag values.
func ParseReviewMode(from, to, commit string) ReviewMode {
  if commit != "" {
    return ModeCommit
  }
  if from != "" && to != "" {
    return ModeRange
  }
  return ModeWorkspace
}

目前ocr支持三种review模式，指定单次commit，指定范围commit，或者针对当前项目。

1
2
3
4
5
6
7
8
9
10
func (m ReviewMode) RefValue(toRef, commit string) (string, bool) {
  switch m {
  case ModeRange:
    return toRef, true
  case ModeCommit:
    return commit, true
  default:
    return "", false
  }
}

RefValue没啥说的，最后整合成为FileReader结构传入buildToolRegistry

1
2
3
4
5
6
7
8
9
func buildToolRegistry(collector *tool.CommentCollector, fr *tool.FileReader) *tool.Registry {
  reg := tool.NewRegistry()
  reg.Register(tool.NewFileRead(fr))
  reg.Register(tool.NewFileFind(fr))
  reg.Register(tool.NewFileReadDiff(tool.DiffMap{}))
  reg.Register(tool.NewCodeSearch(fr))
  reg.Register(&tool.CodeCommentProvider{Collector: collector})
  return reg
}

内置5个工具，代码位于internal/tool/：

  1. FileRead — 通过git或者文件系统读取文件内容
  2. FileFind — 通过git ls查找文件
  3. FileReadDiff — 读取指定文件的diff
  4. CodeSearch — grep代码内容搜索
  5. CodeComment — 提交审查评论
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
func (r *Registry) Register(p Provider) {
  if r.frozen {
    panic("tool: Register called on frozen registry")
  }
  r.providers[p.Tool().name] = p
}
type Tool struct {
  name string
}
type Provider interface {
  // Tool returns which tool this provider implements.
  Tool() Tool
  // Execute runs the tool with the given arguments and returns the result string.
  Execute(ctx context.Context, args map[string]any) (string, error)
}

Execute即是工具的入口函数，其执行结果会放入字符串返回给llm。

initMCPClients

1
2
3
4
5
6
7
8
9
10
func runReview(args []string) error {
  // ...
  mcpClients := initMCPClients(context.Background(), rt.AppCfg, tools, cc.RepoDir, Version)
  defer func() {
    for _, mc := range mcpClients {
      if err := mc.Close(); err != nil {
        fmt.Fprintf(os.Stderr, "[ocr] WARNING: failed to close MCP server %q: %v\n", mc.Name(), err)
      }
    }
  }()

初始化MCP客户端，如果配置有命令行的mcp server也会启动

CollectToolDefs

1
2
3
4
5
func runReview(args []string) error {
  // ...
  mcpToolDefs := mcp.CollectToolDefs(mcpClients, tools)
  rt.PlanToolDefs = append(rt.PlanToolDefs, mcpToolDefs...)
  rt.MainToolDefs = append(rt.MainToolDefs, mcpToolDefs...)

这里将mcp工具转化为了tool的定义并合并入了llmRuntime的PlanToolDefs和MainToolDefs中。

agent.New

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
func runReview(args []string) error {
  // ...
  ag := agent.New(agent.Args{
    RepoDir:               cc.RepoDir,
    From:                  opts.from,
    To:                    opts.to,
    Commit:                opts.commit,
    ReviewMode:            reviewModeFromOptions(opts),
    Template:              *cc.Template,
    SystemRule:            cc.Resolver,
    FileFilter:            cc.FileFilter,
    LLMClient:             rt.Client,
    Tools:                 tools,
    PlanToolDefs:          rt.PlanToolDefs,
    MainToolDefs:          rt.MainToolDefs,
    CommentCollector:      rt.Collector,
    CommentWorkerPool:     agent.NewCommentWorkerPool(opts.concurrency),
    MaxConcurrency:        opts.concurrency,
    ConcurrentTaskTimeout: opts.perFileTimeout,
    Model:                 rt.Model,
    Background:            opts.background,
    GitRunner:             cc.GitRunner,
    Resume:                resumeState,
  })

构造Agent结构

ag.Run

1
2
3
4
5
6
7
8
9
10
11
func runReview(args []string) error {
  // ...
  comments, err := ag.Run(ctx)
  if err != nil {
    span.SetStatus(codes.Error, err.Error())
    span.RecordError(err)
    if id := ag.SessionID(); id != "" {
      fmt.Fprintf(os.Stderr, "[ocr] Session: %s (retry with: --resume %s)\n", id, id)
    }
    return fmt.Errorf("review failed: %w", err)
  }
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
func (a *Agent) Run(ctx context.Context) ([]model.LlmComment, error) {
  // Step 1: Parse diffs
  if err := a.loadDiffs(ctx); err != nil {
    diffSpan.End()
    return nil, fmt.Errorf("load diffs: %w", err)
  }
  // Build the read-only DiffMap from ALL parsed diffs (before filtering)
  // so the LLM can query diffs of related but filtered-out files.
  a.injectDiffMap()
  a.args.Tools.Freeze()

  totalChanged := len(a.diffs)
  reviewCount := a.countReviewable(a.diffs)
  fmt.Fprintf(stdout.Writer(), "[ocr] %d file(s) changed, reviewing %d in %s\n", totalChanged, reviewCount, a.args.RepoDir)

  a.diffs = a.filterDiffs(a.diffs)

  // Step 2: Dispatch per-file subtasks concurrently
  comments, err := a.dispatchSubtasks(ctx)
  if len(comments) > 0 {
    telemetry.RecordCommentsGenerated(ctx, int64(len(comments)))
  }
  a.session.Finalize()
  return comments, err
} // 以上代码我省略了telemetry上报

loadDiffs

1
2
3
4
5
6
7
8
9
10
11
12
13
func (a *Agent) loadDiffs(ctx context.Context) error {
  var provider *diff.Provider

  switch {
  case a.args.Commit != "":
    provider = diff.NewCommitProvider(a.args.RepoDir, a.args.Commit, a.args.GitRunner)
  case a.args.From != "" && a.args.To != "":
    provider = diff.NewProvider(a.args.RepoDir, a.args.From, a.args.To, a.args.GitRunner)
  default:
    provider = diff.NewWorkspaceProvider(a.args.RepoDir, a.args.GitRunner)
  }

  parsed, err := provider.GetDiff(ctx)

根据当前场景执行git命令（GitRunner来自agent亦即commonContext）获取文件diff

injectDiffMap

将此次review的diff注入回FileReadDiff工具里，供下次FileReadDiff工具调用时使用；注入后，a.args.Tools.Freeze()工具被Freeze

countReviewable && filterDiffs

根据用户配置的规则来决定是否review文件，countReviewable只计算数量，filterDiffs剔除不需要review的文件

dispatchSubtasks

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
func (a *Agent) dispatchSubtasks(ctx context.Context) ([]model.LlmComment, error) {
  // Pre-filter: discard diffs whose diff content alone exceeds 80% of the token threshold.
  a.diffs = a.filterLargeDiffs(a.diffs)
  // 使用本地分词器对diff进行统计，如果大于模型上下文80%则抛弃

  toDispatch := a.applyResume(a.diffs)
  // 剔除上次中断之后已经review但未变化的文件

  var wg sync.WaitGroup

  concurrency := a.args.MaxConcurrency
  if concurrency <= 0 {
    concurrency = 8
  }

  sem := make(chan struct{}, concurrency)
  timeout := time.Duration(a.args.ConcurrentTaskTimeout) * time.Minute

  var dispatched int64
  for i := range toDispatch {
    if toDispatch[i].IsDeleted {
      continue
    }
    dispatched++
    wg.Add(1)
    sem <- struct{}{} // acquire semaphore

    go func(d model.Diff) {
      fingerprint := reviewItemFingerprint(a.reviewMode(), d)
      defer wg.Done()
      defer func() { <-sem }() // release
      defer func() {
        if r := recover(); r != nil {
          atomic.AddInt64(&a.subtaskFailed, 1)
          a.session.RecordReviewItemFailed(d.NewPath, d.OldPath, d.NewPath, fingerprint, fmt.Sprintf("panic: %v", r))
          fmt.Fprintf(stdout.Writer(), "[ocr] Subtask panic for %s: %v\n%s\n", d.NewPath, r, debug.Stack())
          a.recordWarning("subtask_error", d.NewPath, fmt.Sprintf("panic: %v", r))
        }
      }()

      var fileCtx context.Context
      var cancel context.CancelFunc
      if timeout > 0 {
        fileCtx, cancel = context.WithTimeout(ctx, timeout)
        defer cancel()
      } else {
        fileCtx = ctx
      }

      completed, skipReason, err := a.executeSubtask(fileCtx, d)
      if err != nil {
        //...
      }
      if !completed {
        if skipReason != "" {
          a.session.RecordReviewItemFailed(d.NewPath, d.OldPath, d.NewPath, fingerprint, skipReason)
        }
        return
      }
      comments := a.args.CommentCollector.CommentsForPath(d.NewPath)
      a.session.RecordReviewItemDone(d.NewPath, d.OldPath, d.NewPath, fingerprint, comments)
    }(toDispatch[i])
  }

  wg.Wait()

  if dispatched == 0 {
    return a.args.CommentCollector.Comments(), nil
  }

  // All subtasks finished — collect comments from the global collector once.
  if a.args.CommentWorkerPool != nil {
    a.args.CommentWorkerPool.Await()
  }

  failed := atomic.LoadInt64(&a.subtaskFailed)
  if failed > 0 && failed == dispatched {
    return nil, fmt.Errorf("all %d file review(s) failed — check your LLM configuration and API key", dispatched)
  }

  return a.args.CommentCollector.Comments(), nil
}

dispatchSubtasks内给每个待review文件建立gorotine

executeSubtask
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
func (a *Agent) executeSubtask(ctx context.Context, d model.Diff) (bool, string, error) {
  newPath := d.NewPath

  // Build change-files list excluding current file
  changeFilesExcludingCurrent := a.buildChangeFilesExcept(newPath)
  // 得到除当前文件以外的其他文件改动背景

  rule := a.resolveSystemRule(strings.ToLower(newPath))
  // 得到当前文件路径的相关审查规则

  threshold := a.args.Template.PlanModeLineThreshold
  changeLines := d.Insertions + d.Deletions

  // Phase 1: Plan (skip when changes are below threshold)
  var planResult string
  if a.args.Template.PlanTask != nil && len(a.args.Template.PlanTask.Messages) > 0 && threshold > 0 && changeLines < int64(threshold) {
    fmt.Fprintf(stdout.Writer(), "[ocr] Skipping plan phase for %s (%d lines < threshold %d)\n", newPath, changeLines, threshold)
  } else if a.args.Template.PlanTask != nil && len(a.args.Template.PlanTask.Messages) > 0 {
    var err error
    planResult, err = a.executePlanPhase(ctx, newPath, d.Diff, changeFilesExcludingCurrent, rule)
    if err != nil {
      fmt.Fprintf(stdout.Writer(), "[ocr] Plan phase failed for %s: %v (continuing without plan)\n", newPath, err)
      planResult = ""
    }
  }

  // Phase 2: Main task loop
  if len(a.args.Template.MainTask.Messages) == 0 {
    return false, "", fmt.Errorf("main_task.messages is empty in template")
  }

  rawMsgs := a.args.Template.MainTask.Messages
  messages := make([]llm.Message, 0, len(rawMsgs))
  for _, m := range rawMsgs {
    content := m.Content
    content = strings.ReplaceAll(content, "{{current_system_date_time}}", a.currentDate)
    content = strings.ReplaceAll(content, "{{current_file_path}}", newPath)
    content = strings.ReplaceAll(content, "{{system_rule}}", rule)
    content = strings.ReplaceAll(content, "{{change_files}}", changeFilesExcludingCurrent)
    content = strings.ReplaceAll(content, "{{diff}}", d.Diff)
    content = strings.ReplaceAll(content, "{{requirement_background}}", a.args.Background)
    if planResult == "" {
      content = stripEmptyPlanBlock(content)
    }
    content = strings.ReplaceAll(content, "{{plan_guidance}}", planResult)
    messages = append(messages, llm.NewTextMessage(m.Role, content))
  }

  tokenCount := llmloop.CountMessagesTokens(messages)
  maxAllowed := a.args.Template.MaxTokens
  tokenLimit := llmloop.PromptTokenLimit(maxAllowed)
  if tokenCount > tokenLimit {
    msg := fmt.Sprintf("prompt tokens (%d) exceed %d%% of max_tokens(%d)", tokenCount, 80, maxAllowed)
    fmt.Fprintf(stdout.Writer(), "[ocr] WARNING: %s for %s\n", msg, newPath)
    a.recordWarning("token_threshold_exceeded", newPath, msg)
    return false, msg, nil
  }

  mainCompleted, err := func() (bool, error) {
    completed, err := a.runner.RunPerFile(ctx, messages, newPath)
    if err != nil {
      mainSpan.SetStatus(codes.Error, err.Error())
      mainSpan.RecordError(err)
      return false, err
    }
    return completed, nil
  }()
  if err == nil {
    if a.args.CommentWorkerPool != nil {
      a.args.CommentWorkerPool.AwaitKey(newPath)
    }
    a.executeReviewFilter(ctx, d, newPath)
  }
  if err != nil {
    return false, "", err
  }
  if !mainCompleted {
    return false, "main_task did not complete before stopping", nil
  }
  return true, "", nil
}
executePlanPhase
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
func (a *Agent) executePlanPhase(ctx context.Context, newPath, rawDiff, changeFiles, rule string) (string, error) {
  pt := a.args.Template.PlanTask
  messages := make([]llm.Message, 0, len(pt.Messages))
  for _, m := range pt.Messages {
    content := m.Content
    content = strings.ReplaceAll(content, "{{current_system_date_time}}", a.currentDate)
    content = strings.ReplaceAll(content, "{{current_file_path}}", newPath)
    content = strings.ReplaceAll(content, "{{system_rule}}", rule)
    content = strings.ReplaceAll(content, "{{change_files}}", changeFiles)
    content = strings.ReplaceAll(content, "{{diff}}", rawDiff)
    content = strings.ReplaceAll(content, "{{requirement_background}}", a.args.Background)
    content = strings.ReplaceAll(content, "{{plan_tools}}", formatToolDefs(a.args.PlanToolDefs))
    messages = append(messages, llm.NewTextMessage(m.Role, content))
  }

  fs := a.session.GetOrCreateFileSession(newPath)
  rec := fs.AppendTaskRecord(session.PlanTask, messages)
  startTime := time.Now()

  resp, err := a.args.LLMClient.CompletionsWithCtx(ctx, llm.ChatRequest{
    Model:     a.args.Model,
    Messages:  messages,
    MaxTokens: a.args.Template.MaxTokens,
  })
  duration := time.Since(startTime)
  if err != nil {
    rec.SetError(err, duration)
    span.SetStatus(codes.Error, err.Error())
    span.RecordError(err)
    return "", fmt.Errorf("plan request: %w", err)
  }
  var totalTokens int64
  if resp.Usage != nil {
    totalTokens = resp.Usage.TotalTokens
  }
  rec.SetResponse(resp, duration)
  a.runner.RecordUsage(resp.Usage)
  fmt.Fprintf(stdout.Writer(), "[ocr] Plan completed for %s\n", newPath)
  return resp.Content(), nil
}

与大模型交互执行提示词模板中的PlanTask部分，plantask的主要功能是review的规划。

RunPerFile
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
func (r *Runner) RunPerFile(ctx context.Context, messages []llm.Message, newPath string) (bool, error) {
  toolReqCount := r.deps.Template.MaxToolRequestTimes
  const maxConsecutiveEmptyRounds = 3
  consecutiveEmptyRounds := 0
  sessionID := uuid.NewString()

  st := &compressionState{}
  defer r.cancelPendingCompression(st)

  for toolReqCount > 0 {
    select {
    case <-ctx.Done():
      return false, ctx.Err()
    default:
    }

    toolReqCount--

    fs := r.deps.Session.GetOrCreateFileSession(ewPath)
    rec := fs.AppendTaskRecord(session.MainTask, append([]llm.Message(nil), messages...))
    startTime := time.Now()

    resp, err := r.deps.LLMClient.CompletionsWithCtx(ctx, llm.ChatRequest{
      Model:     r.deps.Model,
      Messages:  messages,
      Tools:     r.deps.MainToolDefs,
      MaxTokens: r.deps.Template.MaxTokens,
      SessionID: sessionID,
    }) // LLM补全交互
    duration := time.Since(startTime)
    if err != nil {
      rec.SetError(err, duration)
      return false, fmt.Errorf("LLM completion error: %w", err)
    }
    rec.SetResponse(resp, duration)
    totalTokens := int64(0)
    if resp.Usage != nil {
      totalTokens = resp.Usage.TotalTokens
      atomic.AddInt64(&r.totalInputTokens, resp.Usage.PromptTokens)
      atomic.AddInt64(&r.totalOutputTokens, resp.Usage.CompletionTokens)
      atomic.AddInt64(&r.totalCacheReadTokens, resp.Usage.CacheReadTokens)
      atomic.AddInt64(&r.totalCacheWriteTokens, resp.Usage.CacheWriteTokens)
    }

    content := resp.Content()
    calls := resp.ToolCalls()

    if len(calls) == 0 {
      // 无工具调用时，重试下一轮
      fmt.Fprintf(stdout.Writer(), "[ocr] No tool calls parsed for %s, retrying...\n", newPath)
      messages = append(messages, llm.NewTextMessage("user", "You did not successfully call any tools. Please try again or use task_done if finished."))
      if content != "" {
        messages = append(messages[:len(messages)-1], llm.NewTextMessage("assistant", content), messages[len(messages)-1])
      }
      continue
    }

    var results []tool.ToolCallResult
    taskCompleted := false
    hasValidResult := false

    // 遍历llm认为需要的工具调用进行执行
    for _, call := range calls {
      // 执行工具调用
      cp := r.executeToolCall(ctx, newPath, call, rec)
      if cp.Completed {
        results = append(results, tool.ToolCallResult{
          ToolCallID: call.ID,
          Name:       call.Function.Name,
          Result:     "Task completed successfully.",
        })
        taskCompleted = true
      } else if cp.Data != "" {
        results = append(results, tool.ToolCallResult{
          ToolCallID: call.ID,
          Name:       call.Function.Name,
          Result:     cp.Data,
        })
        hasValidResult = true
      } else {
        results = append(results, tool.ToolCallResult{
          ToolCallID: call.ID,
          Name:       call.Function.Name,
          Result:     "Error: Tool execution returned no result.",
        })
      }
    }

    // 任务完成，llm循环结束
    if taskCompleted {
      return true, nil
    }
    // 没有有效的工具调用结果，重试
    if !hasValidResult {
      consecutiveEmptyRounds++
      if consecutiveEmptyRounds >= maxConsecutiveEmptyRounds {
        fmt.Fprintf(stdout.Writer(), "[ocr] Too many empty retries for %s, stopping.\n", newPath)
        break
      }
      fmt.Fprintf(stdout.Writer(), "[ocr] No valid tool results for %s, retrying...\n", newPath)
    } else {
      consecutiveEmptyRounds = 0
    }

    // 
    succeed := r.addNextMessage(ctx, content, calls, results, &messages, newPath, st)
    if !succeed {
      fmt.Fprintf(stdout.Writer(), "[ocr] Context compression exceeded threshold for %s, stopping.\n", newPath)
      break
    }
  }

  if toolReqCount <= 0 {
    fmt.Fprintf(stdout.Writer(), "[ocr] Max tool requests reached for %s.\n", newPath)
  }
  return false, nil
}

review的LLM主循环，MainTask的提示词在executeSubtask中被replace之后被放入messages，每次循环会往该messages中添加，addNextMessage会触发上下文压缩

addNextMessage

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
func (r *Runner) addNextMessage(ctx context.Context, assistantContent string, toolCalls []llm.ToolCall, results []tool.ToolCallResult, messages *[]llm.Message, filePath string, st *compressionState) bool {
  maxAllowed := r.deps.Template.MaxTokens
  softLimit := int(float64(maxAllowed) * tokenSoftThreshold)
  warnLimit := PromptTokenLimit(maxAllowed)

  r.tryApplyPendingCompression(st, messages)
  // 将上一轮异步压缩triggerAsyncCompression的message替换当前

  if CountMessagesTokens(*messages) > warnLimit {
    // warnLimit默认为80%的maxTokens，如果超过则放弃异步压缩，先同步压缩
    r.cancelPendingCompression(st)
    var err error
    if *messages, err = r.runCompression(ctx, *messages, filePath); err != nil {
      fmt.Fprintf(stdout.Writer(), "[ocr] Memory compression failed: %v\n", err)
    }
  }

  // 追加本轮对话消息
  if len(toolCalls) > 0 {
    *messages = append(*messages, llm.NewToolCallMessage(assistantContent, toolCalls))
  } else if assistantContent != "" {
    *messages = append(*messages, llm.NewTextMessage("assistant", assistantContent))
  }

  // 追加工具调用结果
  for _, rs := range results {
    *messages = append(*messages, llm.NewToolResultMessage(rs.ToolCallID, rs.Result))
  }

  // 如果当前消息数量超过80%的maxTokens，则再同步压缩一次
  finalCount := CountMessagesTokens(*messages)
  if finalCount > warnLimit {
    r.cancelPendingCompression(st)
    var err error
    if *messages, err = r.runCompression(ctx, *messages, filePath); err != nil {
      fmt.Fprintf(stdout.Writer(), "[ocr] Memory compression failed: %v\n", err)
    }
    finalCount = CountMessagesTokens(*messages)
  }

  // 如果当前消息数量超过softLimit（默认值60%），小于warnLimit（默认值80%），则触发异步压缩，异步压缩为gorotine的runCompression
  if finalCount > softLimit && finalCount < warnLimit {
    r.triggerAsyncCompression(ctx, st, *messages, filePath)
  }

  return finalCount < warnLimit
}

runCompression

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
func (r *Runner) runCompression(ctx context.Context, msgs []llm.Message, filePath string) ([]llm.Message, error) {
  if len(r.deps.Template.MemoryCompressionTask.Messages) == 0 || len(msgs) <= 2 {
    return msgs[:min(len(msgs), 2)], nil
  }

  // 将messages进行分区，frozen区为提示词模板内容，compress为中间N轮对话，active为最近K轮
  // fronzen不压缩，active的计算按照剩余token最多能装满的值区分，使用本地分词器估算token
  part := partitionMessages(msgs, r.deps.Template.MaxTokens, 0)
  if part.compressEnd <= part.frozenEnd {
    return msgs, nil
  }

  // 将compress区转换为xml格式
  contextXML := buildMessageXML(msgs[part.frozenEnd:part.compressEnd])

  // 根据压缩提示词模板进行替换
  // 模板见internal/config/template/prompts/memory_compression_task_system.md && memory_compression_task_user.md
  compressionMsgs := make([]llm.Message, 0, len(r.deps.Template.MemoryCompressionTask.Messages))
  for _, m := range r.deps.Template.MemoryCompressionTask.Messages {
    content := strings.ReplaceAll(m.Content, "{{context}}", contextXML)
    compressionMsgs = append(compressionMsgs, llm.NewTextMessage(m.Role, content))
  }

  // 调用LLM进行压缩
  startTime := time.Now()
  resp, err := r.deps.LLMClient.CompletionsWithCtx(ctx, llm.ChatRequest{
    Model:     r.deps.Model,
    Messages:  compressionMsgs,
    MaxTokens: r.deps.Template.MaxTokens,
  })
  duration := time.Since(startTime)

  fs := r.deps.Session.GetOrCreateFileSession(filePath)
  rec := fs.AppendTaskRecord(session.MemoryCompressionTask, compressionMsgs)
  if err != nil {
    rec.SetError(err, duration)
    return msgs, fmt.Errorf("memory compression: %w", err)
  }
  rec.SetResponse(resp, duration)
  if resp.Usage != nil {
    atomic.AddInt64(&r.totalInputTokens, resp.Usage.PromptTokens)
    atomic.AddInt64(&r.totalOutputTokens, resp.Usage.CompletionTokens)
    atomic.AddInt64(&r.totalCacheReadTokens, resp.Usage.CacheReadTokens)
    atomic.AddInt64(&r.totalCacheWriteTokens, resp.Usage.CacheWriteTokens)
  }

  // 去除结果markdown中代码反引号
  rawSummary := stripMarkdownFences(resp.Content())
  if rawSummary == "" {
    return msgs, nil
  }

  // 重建新的messages，前2为fronzen区
  rebuilt := make([]llm.Message, 2)
  copy(rebuilt, msgs[:2])

  // 压缩后的文本被添加入user提示词（rebuilt[1]）的结尾
  userMsg := rebuilt[1]
  currentText := userMsg.ExtractText()
  rebuilt[1] = llm.NewTextMessage(userMsg.Role, currentText+"\n\n<previous_review_summary>\n"+rawSummary+"\n</previous_review_summary>")

  // active区补充
  for i := part.compressEnd; i < len(msgs); i++ {
    rebuilt = append(rebuilt, msgs[i])
  }

  return rebuilt, nil
}
executeReviewFilter
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
func (a *Agent) executeReviewFilter(ctx context.Context, d model.Diff, newPath string) {
  // review filter模板
  ft := a.args.Template.ReviewFilterTask
  if ft == nil || len(ft.Messages) == 0 {
    return
  }

  // llm对于文件的review结果在comment中
  comments := a.args.CommentCollector.CommentsForPath(newPath)
  if len(comments) == 0 {
    return
  }

  // 将comment内容、原始代码拼为json
  commentsJSON := buildFilterCommentsJSON(comments)

  messages := make([]llm.Message, 0, len(ft.Messages))
  for _, m := range ft.Messages {
    content := m.Content
    content = strings.ReplaceAll(content, "{{path}}", newPath)
    content = strings.ReplaceAll(content, "{{diff}}", d.Diff)
    content = strings.ReplaceAll(content, "{{comments}}", commentsJSON)
    // 构建给llm的messages
    messages = append(messages, llm.NewTextMessage(m.Role, content))
  }

  if ft.Timeout > 0 {
    var cancel context.CancelFunc
    ctx, cancel = context.WithTimeout(ctx, time.Duration(ft.Timeout)*time.Second)
    defer cancel()
  }

  fs := a.session.GetOrCreateFileSession(newPath)
  rec := fs.AppendTaskRecord(session.ReviewFilterTask, messages)
  startTime := time.Now()

  _, llmSpan := telemetry.StartLLMSpan(ctx, a.args.Model)
  // 再次调用大模型来review 之前review的结果
  resp, err := a.args.LLMClient.CompletionsWithCtx(ctx, llm.ChatRequest{
    Model:     a.args.Model,
    Messages:  messages,
    MaxTokens: a.args.Template.MaxTokens,
  })
  duration := time.Since(startTime)

  var totalTokens int64
  if resp.Usage != nil {
    totalTokens = resp.Usage.TotalTokens
  }
  telemetry.RecordLLMResult(llmSpan, duration, totalTokens, nil)
  llmSpan.End()
  rec.SetResponse(resp, duration)
  a.runner.RecordUsage(resp.Usage)

  // 将llm输出转为map[int]struct{}，key为对应comment的index
  indices := parseFilterResponse(resp.Content(), len(comments))
  telemetry.SetAttr(span, "comments.filtered", len(indices))
  if len(indices) == 0 {
    return
  }

  // 根据过滤后的index删除指定comment
  a.args.CommentCollector.RemoveByPathAndIndices(newPath, indices)
  fmt.Fprintf(stdout.Writer(), "[ocr] Review filter removed %d comment(s) for %s\n", len(indices), newPath)
}
通过再一次的llm对话（但只有一次）来过滤掉有问题的comment

emitRunResult

1
2
3
4
func runReview(args []string) error {
  // ...
  return emitRunResult(ctx, ag, comments, startTime, opts.outputFormat, opts.audience, q)
}

对review结果（comment）进行格式化并输出

总结

可以看出，相比我之前使用的skill reivew方式，ocr有着更好的一套工程化方法（plan->并发review->review再过滤）。
后续使用上如果想到啥会再更新这篇，那么ocr的源码解析先到此了。