跳到主要内容

数据截至 (上游 commit ae57a2357745)

第 2 章 · 工具协议与执行循环

本章讲:在没有 function calling 的前提下,模型说的一段话怎么变成机器上真实跑起来的进程,跑挂了又怎么让模型自己修。


2.1 它要解决的小问题

Agent 要用工具,就得有个「协议」让模型表达「我要调用 X,参数是 Y」。主流答案是 OpenAI 的 function calling:模型输出结构化 JSON。

但 AgenticSeek 的目标是本地小模型,而小模型输出严格 JSON 的失败率不低——多一个逗号、少一个引号、外面裹一层解释文字,解析就崩。

项目的答案很朴素:

模型最会写的东西是代码块,那就把代码块当协议。


2.2 协议长什么样

每个工具认领一个围栏标签(Tools.tag),模型只要写出带该标签的 markdown 代码块,系统就会执行它。

下面是 prompts/base/coder_agent.txt 里教写法的那一段节选原文(未改写,原文即英文):

You can execute bash command using the bash tag :
```bash
#!/usr/bin/env bash
ls -la # example
```

You can execute python using the python tag
```python
print("hey")
```

You can execute go using the go tag, as you can see adding :filename will save the file.
```go:hello.go
package main

func main() {
fmt.Println("hello")
}
```

三段依次是:跑 bash、跑 python、跑 go 并顺便存成 hello.go。注意第三个——围栏行上的 :hello.go。它不是语法糖装饰,是真的会存盘(见 2.3、2.4)。

现有的标签分配:

标签工具类干什么归谁用
bashBashInterpreter跑 shellcoder / file
pythonPyInterpreter跑 Pythoncoder
c / go / javaCInterpreter / GoInterpreter / JavaInterpreter编译后运行coder
file_finderFileFinder递归找文件、读内容coder / file
web_searchsearxSearch查 SearxNGbrowser
jsonTools(裸基类,只解析不执行)接住规划 JSONplanner
mcp_finderMCP_finder查 Smithery 注册表mcp(未启用)
flight_searchFlightSearchSerpApi 航班查询无(只被 casual 文件 import,从未接入)

两行需要单独说明:

  • json 那一行是个巧妙复用:PlannerAgent 直接 Tools() 造一个基类实例,只把 tag 设成 "json",借用它的块解析能力,从不调用 executesources/agents/planner_agent.py:20-23)。
  • flight_search只 import 未接入sources/agents/casual_agent.py:6from sources.tools.flightSearch import FlightSearch,但 CasualAgent.__init__self.tools = {} 是空 dict,全仓库唯一一处 FlightSearch() 构造在 sources/tools/flightSearch.py:86__main__ 自测块里。所以这个标签在运行时永远不会被任何 agent 认领。

2.3 块解析:load_exec_block

核心就一个函数(sources/tools/tools.py:150-200)。逻辑是纯字符串扫描,没有正则、没有 markdown parser:

找 "```<tag>" 的位置 ─┐
│ 找不到 → 返回 (None, None),本工具没活
v
记下这行开头到围栏之间的空白 = leading_whitespace
v
往后找第一个 "```" 作为结束
v
取中间内容,逐行剥掉 leading_whitespace ← 反缩进
v
围栏行标签后面还有 ":" ? → 冒号后面当 save_path,这截残余丢弃
v
塞进 code_blocks,从结束位置继续找下一个

为什么要反缩进

模型经常把代码块嵌在列表或引用里,于是围栏和代码整体带 4 个空格。直接执行的话 Python 立刻 IndentationError。这段做的事是:围栏行前面有多少空白,就把块内每一行都剥掉多少sources/tools/tools.py:176-191):

line_start = llm_text.rfind('\n', 0, start_pos)+1
leading_whitespace = llm_text[line_start:start_pos]
...
for line in lines:
if line.startswith(leading_whitespace):
processed_lines.append(line[len(leading_whitespace):])
else:
processed_lines.append(line)

tests/test_tools_parsing.py:85test_load_exec_block_with_indentation)专门盯着这个行为。

save_path 认的是围栏行的尾巴,不是代码第一行

存盘判定这三行很容易看错(sources/tools/tools.py:193-195):

if ':' in content.split('\n')[0]:
save_path = content.split('\n')[0].split(':')[1]
content = content[content.find('\n')+1:]

关键在 content 是从哪里开始截的(sources/tools/tools.py:182):

content = llm_text[start_pos + len(start_tag):end_pos]

start_tag```python 这一整串,所以 content 的第一行是围栏行上标签后面剩下的那点尾巴——正常多行写法下它是空串,而不是代码的第一行。于是:

模型写的content 首行结果
```python 换行后 for i in x:""save_path = None,代码一行不少
```python:main.py":main.py"save_path = "main.py",围栏行残余被丢弃

也就是说,代码正文里的冒号(for i in x:d = {"a": 1})既不会触发存盘,也不会被削掉。仓库自带测试正是反向断言这一点:tests/test_tools_parsing.py:69test_load_exec_block_with_save_path)把 save_path: test_file.py 写在块内部第一行,断言返回的 save_path 仍是 None、那一行原样留在代码里。

真正需要留意的边界有两处:

  • 整块被挤成一行时(围栏、代码、结束围栏同处一行),首行就是代码本身,里面的冒号会被当成路径。此时 content.find('\n') 返回 -1、content[0:] 等于原串,所以代码不会被削,只是白白得到一个垃圾 save_path
  • save_path 是循环外的单个变量,同一工具的多个块共用最后一次赋值;而 save_block 又在 for 里用 'w' 打开同一个文件(sources/tools/tools.py:123-125),所以一次回复里多个块要存盘时,文件里只会剩最后一块

2.4 执行循环:execute_modules

拿到模型回复后,Agent.execute_modules() 遍历自己所有工具,逐个问「这段文本里有你的块吗」(sources/agents/agent.py:255-285)。

模型回复 answer
|
v
以 "```" 开头? → 前面补一句 "I will execute:\n"
| (下游 show_answer 假设块前有文字)
v
for name, tool in self.tools.items():
|
├─ blocks, save_path = tool.load_exec_block(answer)
├─ blocks 为 None → 跳过这个工具
|
└─ 对每个 block:
展示 → tool.execute([block])
→ tool.interpreter_feedback(output) 包装成 [success]/[failure]
→ tool.execution_failure_check(output) 判成败
→ 存进 blocks_result
→ 失败:memory.push('user', feedback) 并【立即 return False】
全部成功 → memory.push('user', feedback)
→ save_path 有值就 tool.save_block(...)

三个值得记住的设计

① 失败即刹车。 一个块失败就 return False, feedback后面的块和后面的工具都不再执行sources/agents/agent.py:278-281)。代价是「先建目录、再写文件」这种链式操作,前一步失败后一步也不会瞎跑。

② 反馈伪装成用户消息。 无论成败,喂回 memory 的角色都是 'user'

self.memory.push('user', feedback)

对模型来说,「解释器骂你了」和「用户骂你了」是同一种输入。好处是不需要 tool/function 这个消息角色,任何 chat 接口(包括最简陋的本地服务)都能吃。

③ 只喂最后一个 feedback。 循环里 feedback 每轮被覆盖,成功路径下只把最后一个块的反馈推进 memory。多个块都成功时,前面几个的输出模型是看不到的。


2.5 答案里的代码块去哪了:remove_blocksblock:N

执行完之后,答案文本里的代码块会被替换成占位符sources/agents/agent.py:226-245):

if tag in line and not in_block:
in_block = True
continue
if not in_block:
post_lines.append(line)
if tag in line:
in_block = False
post_lines.append(f"block:{block_idx}")
block_idx += 1

于是「我来列一下文件」+ 一段 python 代码,变成:

我来列一下文件
block:0

展示时再由 show_answer() 按索引把执行结果插回原位(sources/agents/agent.py:210-224),API 侧则把 blocks_result 序列化成 JSON 交给前端渲染(api.py:263)。

这套占位符的价值:答案文本和执行结果解耦了——文字可以进 memory、可以朗读,而沉重的代码与输出单独存在 blocks_result 里。


2.6 各解释器怎么跑

语言执行方式超时特殊处理
Pythonsubprocess.run([sys.executable, "-c", code], cwd=work_dir)300s交互式代码开跑前就拒绝
Bashsubprocess.Popen(command, shell=True, cwd=work_dir)300s拦截「用 bash 去跑别的语言」
C临时目录里 gcc 编译再运行编译 60s / 运行 120s
Go临时目录里 go buildGO111MODULE=off)再运行编译 10s / 运行 10s
Java临时目录里 javac + java编译 10s / 运行 10s

(超时值见 sources/tools/C_Interpreter.py:45/56GoInterpreter.py:45/57JavaInterpreter.py:43/54。)

Python:先拒绝再执行

PyInterpreter 有一份「注定跑不通」的模式表(sources/tools/PyInterpreter.py:17-20):

INTERACTIVE_PATTERNS = [
(re.compile(r"^\s*(?:import|from)\s+[^\n]*\bcurses\b", re.MULTILINE), "curses"),
(re.compile(r"(?<![\w.])(?<!def\s)input\s*\("), "input()"),
]

命中就直接返回一段可操作的拒绝话术,而不是让它跑到 EOF 报错(refuse_interactive_codesources/tools/PyInterpreter.py:28-41):

“…requires an interactive terminal, but code runs headless in a sandbox with no terminal attached. Rewrite the code without input(): take values from variables in the code and print results to stdout.”

妙在哪:错误信息本身就是给模型的修改指令。让 input() 真跑一遍只会得到一句 EOFError,模型看了未必知道该怎么改;这段话直接告诉它「把值写进变量、结果打到 stdout」。

那个负向前瞻 (?<!def\s)(?<![\w.]) 也有讲究——obj.input(...)def input(...) 不该被误伤,tests/test_interpreters.py:138test_input_method_call_is_not_refused)盯着这条。

Bash:拦截「套娃执行」

模型常犯的毛病是写完 Python 又补一句 python main.py。但代码块本来就会被自动执行,再跑一次就是重复。language_bash_attempt 扫命令里有没有 python/gcc/go/java 等开头的词,有就静默跳过这条命令sources/tools/BashInterpreter.py:23-3354-55):

if self.language_bash_attempt(command) and self.allow_language_exec_bash == False:
continue

allow_language_exec_bash 有 setter,但代码里没有任何地方把它设为 True,所以实际总是拦。

成败判定:关键词匹配,不看退出码

所有解释器的 execution_failure_check 都是在输出文本里正则搜错误关键词sources/tools/BashInterpreter.py:88-123):

error_patterns = [r"expected", r"errno", r"failed", r"invalid", ..., r"not found", r"missing", ...]

这条设计简单,但误报是必然的:程序正常打印 “file not found” 会被判失败,ls 输出里恰好有个叫 missing.txt 的文件也会。Bash 的关键词表比 Python 的长得多——bash 侧 26 条(sources/tools/BashInterpreter.py:93-118),Python 侧 11 条(sources/tools/PyInterpreter.py:93-105),所以 bash 的误报面更大。


2.7 agent 的自愈循环

各 agent 的 process() 结构一致,以 CoderAgent 为例(sources/agents/code_agent.py:46-86):

prompt 加上系统信息(OS / Python 版本 / 必须存到哪个目录)
v
memory.push('user', prompt)
v
┌──── while attempt < 5 且未被 stop ────────────────┐
│ llm_request() │
│ 答案含 REQUEST_CLARIFICATION?→ 直接返回问用户 │
│ 答案不含 ``` ?→ 纯文字,break │
│ execute_modules(answer) │
│ remove_blocks(answer) │
│ 成功 且 最后一个工具不是 bash → break │
│ 否则:打印失败、attempt += 1,回到循环顶 │
└────────────────────────────────────────────────────┘
v
attempt 用满 5 次 → 返回道歉话术

自愈是怎么发生的:失败时 execute_modules 已经把 [failure] Error in execution:\n<stderr> 推进 memory 了,下一轮 llm_request() 读到的历史里就带着报错,模型自然会改。没有任何显式的「重试 prompt」——全靠 memory 里那条伪装成用户消息的反馈。

bash 的例外分支(一个真实的粗糙点)

if exec_success and self.get_last_tool_type() != "bash":
break

bash 成功也不 break。 意图看起来是让模型能连着下多条 shell 命令,但副作用有两个:

  1. 每轮都会打印 “Execution failure” + “Correcting code...”,哪怕执行是成功的;
  2. 顺利跑满 5 轮之后,函数返回 “I'm sorry, I couldn't find a solution to your problem. How would you like me to proceed ?”(sources/agents/code_agent.py:83-84)——成功的操作报告成失败

tests/test_agent_regressions.py:46test_retry_loop_is_capped)只验证了循环有上限,没验证这条语义。

各 agent 的循环差异

Agent最大轮数退出条件
CasualAgent不循环,问一次就返回
CoderAgent5执行成功且最后工具非 bash / 无代码块 / 请求澄清
FileAgent5exec_success 为真
McpAgent5本轮没产生新 block 就 break
BrowserAgent无固定上限未访问链接耗尽 / 模型说 REQUEST_EXIT

2.8 异步是怎么做的(以及为什么这么做)

LLM 调用是同步阻塞的(各 provider 用 requests/openai SDK)。为了不卡住 FastAPI 的事件循环,Agent 每个实例自带一个单线程执行器(sources/agents/agent.py:51160-178):

self.executor = ThreadPoolExecutor(max_workers=1)
...
async def llm_request(self):
loop = asyncio.get_event_loop()
return await loop.run_in_executor(self.executor, self.sync_llm_request)

max_workers=1 是有意的:同一个 agent 的 LLM 调用天然串行,memory 是共享可变状态,并发会写坏。

注意工具执行不在这套异步里——execute_modules 是同步调用的,一个 300 秒超时的 bash 命令会实打实地阻塞事件循环。各 process() 里那些 await asyncio.sleep(0) 是在给事件循环让出机会,好让 /latest_answer 轮询能读到中间状态。


2.9 reasoning 模型的 <think> 怎么处理

面向 DeepSeek-R1 这类会输出思维链的模型,基类做了一刀切分(sources/agents/agent.py:138-158):

def remove_reasoning_text(self, text):
end_idx = text.rfind("</think>")
if end_idx == -1:
return text
return text[end_idx+8:] # 8 == len("</think>")

def extract_reasoning_text(self, text):
start_idx = text.find("<think>")
end_idx = text.rfind("</think>")+8
return text[start_idx:end_idx]

切完之后:

  • 答案(去掉思维链)进 memory,也就是说思维链不占后续上下文
  • 思维链单独存 last_reasoning,透给前端做「展开看推理」(api.py:193)。

rfind 找结束标签是为了应付模型嵌套或重复输出 </think> 的情况。


2.10 本章代码地图

主题文件符号
块解析 + 反缩进 + 存盘路径sources/tools/tools.pyTools.load_exec_blockTools.save_block
参数解析(name=xxxsources/tools/tools.pyTools.get_parameter_value
工具遍历与执行sources/agents/agent.pyAgent.execute_modules
占位符与展示sources/agents/agent.pyAgent.remove_blocksshow_answerraw_answer_blocks
异步包装sources/agents/agent.pyAgent.llm_requestsync_llm_request
思维链切分sources/agents/agent.pyremove_reasoning_textextract_reasoning_text
Python 执行sources/tools/PyInterpreter.pyPyInterpreter.executerefuse_interactive_code
Bash 执行sources/tools/BashInterpreter.pyBashInterpreter.executelanguage_bash_attempt
编译型语言sources/tools/C_Interpreter.pyGoInterpreter.pyJavaInterpreter.pyCInterpreter.executeGoInterpreter.executeJavaInterpreter.execute
文件查找sources/tools/fileFinder.pyFileFinder.executerecursive_search
未接入的工具sources/tools/flightSearch.pyFlightSearch
执行结果模型sources/schemas.pyexecutorResult
coder 循环sources/agents/code_agent.pyCoderAgent.process
块解析测试tests/test_tools_parsing.pytest_load_exec_block_with_indentationtest_load_exec_block_with_save_path