第 3 章 · 01 CodeAgent 与代码执行循环 ★
第 3 章 · 01 CodeAgent 与代码执行循环 ★
本节摘要:本节进入全书第一个高潮——精读
agents.py:1505-1803的CodeAgent,smolagents 的灵魂,agency 等级表 ★★★ 的"代码 agent"。它继承 MultiStepAgent 却只覆写单步语义:LLM 的每一步 action 是一段 ```python 代码块;parse_code_blobs把代码从输出文本里抠出来;代码送入LocalPythonExecutor的evaluate_python_code执行(自研 AST 解释器,第 6 章精读);工具不是"被调度的外部服务",而是被send_tools注入解释器命名空间的普通 Python 函数——web_search(query=...)在代码里就是一个函数调用;代码里调final_answer(...)即终止任务。本节把这条"LLM → 代码 → 解析 → 执行 → 观察 → 循环"的完整链路逐段拆开,并给出全链路 Mermaid 图。
内容来源:原项目源码
src/smolagents/agents.py(CodeAgent 类)、src/smolagents/utils.py(parse_code_blobs)、src/smolagents/local_python_executor.py(LocalPythonExecutor/send_tools/fix_final_answer_code)。
⚠️ 注意:本地执行器
LocalPythonExecutor不是安全沙箱(源码 docstring 原话 "It is not a security sandbox")。authorized_imports白名单只是"提醒 LLM 别乱 import + 解释器拦截部分危险模块",恶意代码有绕过手段。生产环境执行 LLM 生成的代码必须用executor_type="e2b"/"docker"/"modal"/"blaxel"远程沙箱(第 6 章)。
学习目标
阅读完本节,你应当能够:
- 说出 CodeAgent 相对 MultiStepAgent 新增的构造参数(executor/executor_type/code_block_tags/authorized_imports)。
- 逐段讲解 CodeAgent
_step_stream的四段:生成 → 解析 → 执行 → 观察。 - 解释
parse_code_blobs的三级降级策略(标签匹配 → markdown 匹配 → 整段当代码)。 - 说清"工具 = 解释器命名空间里的 Python 函数"这一设计的实现:
send_tools与run()里的注入时机。 - 解释
final_answer(...)如何终止循环、fix_final_answer_code修的是什么坑。 - 画出代码执行循环的完整链路图。
一、CodeAgent 类:构造参数与执行器选择
agents.py:1505-1585(节选):
1505 class CodeAgent(MultiStepAgent): 1506 """ 1507 In this agent, the tool calls will be formulated by the LLM in code format, 1508 then parsed and executed. 1509 """ 1527 def __init__(self, tools, model, prompt_templates=None, 1532 additional_authorized_imports: list[str] | None = None, 1534 executor: PythonExecutor = None, 1535 executor_type: Literal["local", "blaxel", "e2b", "modal", "docker"] = "local", 1536 executor_kwargs: dict | None = None, 1537 max_print_outputs_length: int | None = None, 1539 use_structured_outputs_internally: bool = False, 1540 code_block_tags: str | tuple[str, str] | None = None, 1541 **kwargs): 1543 self.additional_authorized_imports = additional_authorized_imports or [] 1544 self.authorized_imports = sorted(set(BASE_BUILTIN_MODULES) | set(self.additional_authorized_imports)) 1552 prompt_templates = prompt_templates or yaml.safe_load( 1553 importlib.resources.files("smolagents.prompts").joinpath("code_agent.yaml").read_text() 1554 ) 1556 if isinstance(code_block_tags, str) and not code_block_tags == "markdown": 1557 raise ValueError("Only 'markdown' is supported for a string argument to `code_block_tags`.") 1558 self.code_block_tags = ( 1559 code_block_tags if isinstance(code_block_tags, tuple) 1560 else ("```python", "```") if code_block_tags == "markdown" 1561 else ("<code>", "</code>") 1562 ) 1583 self.executor_type = executor_type 1585 self.python_executor = executor or self.create_python_executor()
关键参数:
| 参数 | 含义 |
|---|---|
additional_authorized_imports |
在默认白名单之外放行的 import(如 "requests"、"numpy.random",支持 "numpy.*" 与通配 "*") |
executor / executor_type |
自定义执行器,或从五种类型选一种,默认 local |
code_block_tags |
代码块边界标签:默认 <code>...</code>;传 "markdown" 用 ```python;也可传自定义二元组 |
use_structured_outputs_internally |
每步用结构化输出(structured_code_agent.yaml,第三节讲) |
max_print_outputs_length |
print 输出的截断长度(默认 50000 字符,防观察爆炸) |
BASE_BUILTIN_MODULES(utils.py:49-61)是默认白名单:collections/datetime/itertools/math/queue/random/re/stat/statistics/time/unicodedata——全是纯计算标准库,没有 os/sys/subprocess/socket。create_python_executor(1598-1618)按 executor_type 实例化:local → LocalPythonExecutor;其余四种 → 对应 Remote 执行器,且远程模式下禁止 managed_agents(1608-1609 显式抛错,因为子 agent 对象没法序列化进远程沙箱)。
注意默认代码块标签是 <code> 而非 markdown 围栏——好处见第二段"stop_sequences 智能裁剪"。
二、_step_stream 四段:生成 → 解析 → 执行 → 观察
第 1 段:生成(agents.py:1646-1700)
1646 memory_messages = self.write_memory_to_messages() # 记忆 → 消息 1651 stop_sequences = ["Observation:", "Calling tools:"] 1652 if self.code_block_tags[1] not in self.code_block_tags[0]: 1653 stop_sequences.append(self.code_block_tags[1]) # 闭标签做停止序列 1677 chat_message: ChatMessage = self.model.generate( 1678 input_messages, stop_sequences=stop_sequences, **additional_args) 1683 output_text = chat_message.content 1693 if output_text and not output_text.strip().endswith(self.code_block_tags[1]): 1694 output_text += self.code_block_tags[1] # 手动补闭标签
和 ToolCallingAgent 一样禁止 LLM 自演 Observation。精妙处在 1652-1654 与 1693-1694:默认标签 <code>/</code> 互不包含,把闭标签加进停止序列,LLM 写完 </code> 就停,省 token;而 markdown 模式下开标签 ```python 本身包含闭标签 ```,若再加停止序列会把代码拦腰截断,所以只在生成后检查补齐(1690 行注释称这还能"示范收尾"引导后续轮次)。
第 2 段:解析(agents.py:1702-1721)
1703 try: 1704 if self._use_structured_outputs_internally: 1705 code_action = json.loads(output_text)["code"] 1706 code_action = extract_code_from_text(code_action, self.code_block_tags) or code_action 1707 else: 1708 code_action = parse_code_blobs(output_text, self.code_block_tags) 1709 code_action = fix_final_answer_code(code_action) 1710 memory_step.code_action = code_action 1711 except Exception as e: 1712 raise AgentParsingError(f"Error in code parsing:\n{e}\nMake sure to provide correct code blobs.", ...) 1715 tool_call = ToolCall(name="python_interpreter", 1717 arguments=code_action, id=f"call_{len(self.memory.steps)}") 1720 yield tool_call 1721 memory_step.tool_calls = [tool_call]
整段代码被包装成一个名为 python_interpreter 的 ToolCall 存档——在记忆的序列化视图里,CodeAgent 每步"只调用了一个工具:Python 解释器"。parse_code_blobs 的三级降级见下节;fix_final_answer_code 见第四节。
第 3 段:执行(agents.py:1723-1750)
1724 self.logger.log_code(title="Executing parsed code:", content=code_action, ...) 1726 code_output = self.python_executor(code_action) # 调 LocalPythonExecutor.__call__ 1727 if len(code_output.logs) > 0: 1733 observation = "Execution logs:\n" + code_output.logs # print 输出 = 观察主体 1734 except Exception as e: 1735 if hasattr(self.python_executor, "state") and "_print_outputs" in ...: 1736 execution_logs = str(self.python_executor.state["_print_outputs"]) 1742 memory_step.observations = "Execution logs:\n" + execution_logs # 崩溃也保留已打印内容 1745 if "Import of " in str(e) and " is not allowed" in str(e): 1746 self.logger.log("[bold red]Warning to user: ... unauthorized import - " 1747 "Consider passing said import under `additional_authorized_imports` ...") 1750 raise AgentExecutionError(error_msg, self.logger)
self.python_executor(code_action) 一行触发 AST 解释执行(细节第 6 章),返回 CodeOutput(output, logs, is_final_answer)。执行抛异常时依然把崩溃前已产生的 print 日志抢救进观察——LLM 下一步能看到"跑到哪一步、打印了什么、死在哪",精准改代码重试。未授权 import 的错误还会额外提示用户检查 additional_authorized_imports——错误信息同时服务 LLM 和人类开发者。
第 4 段:观察回填(agents.py:1752-1764)
1752 truncated_output = truncate_content(str(code_output.output)) # 超长截断(2 万字符) 1753 observation += "Last output from code snippet:\n" + truncated_output 1754 memory_step.observations = observation 1763 memory_step.action_output = code_output.output 1764 yield ActionOutput(output=code_output.output, is_final_answer=code_output.is_final_answer)
观察 = print 日志 + 最后一行表达式的值(截断后)。is_final_answer 为真时 _run_stream 翻标志结束任务。
三、parse_code_blobs:三级降级的代码提取
utils.py:198-251:
212 matches = extract_code_from_text(text, code_block_tags) # ① 按配置标签正则提取 213 if not matches: # Fallback to markdown pattern 214 matches = extract_code_from_text(text, ("```(?:python|py)", "\n```")) # ② markdown 兜底 215 if matches: 216 return matches 218 try: 219 ast.parse(text) # ③ 整段能通过语法解析 → 当作纯代码 220 return text 221 except SyntaxError: 222 pass 224 if "final" in text and "answer" in text: # ④ 针对性错误提示 225 raise ValueError("... It seems like you're trying to return the final answer, you can do it as follows: <code>final_answer(\"...\")</code> ...")
①默认正则 <code>(.*?)</code>(extract_code_from_text,DOTALL 多行);②LLM 不守规矩改用 markdown 围栏时兜住;③有些模型直接裸吐代码不加标签,ast.parse 验证语法合法就整段收下;④全失败时给出含正确示例的报错(下轮 LLM 照抄即可修复)。四个 fallback 层层放宽,把"解析失败率"压到极低——这是代码范式能落地的一等功臣。
四、工具即函数:send_tools 与 final_answer
注入时机在 MultiStepAgent.run()(第 2 章读过的 agents.py:490-492):
490 if getattr(self, "python_executor", None): 491 self.python_executor.send_variables(variables=self.state) 492 self.python_executor.send_tools({**self.tools, **self.managed_agents})
工具与子 agent 合并后一起送进执行器。local_python_executor.py:1688 起:
1760 def send_variables(self, variables: dict[str, Any]): 1761 self.state.update(variables) # additional_args 变量 → 命名空间变量 1763 def send_tools(self, tools: dict[str, Tool]): 1764 # 把 agent 工具、基础 Python 工具、附加函数合并进 static_tools 1764 self.static_tools = {**tools, **BASE_PYTHON_TOOLS.copy(), **self.additional_functions} 1747 def __call__(self, code_action: str) -> CodeOutput: 1748 output, is_final_answer = evaluate_python_code( 1749 code_action, static_tools=self.static_tools, custom_tools=self.custom_tools, 1750 state=self.state, authorized_imports=self.authorized_imports, ...) 1753 return CodeOutput(output=output, logs=str(self.state["_print_outputs"]), is_final_answer=is_final_answer)
于是解释器里同时存在三类"函数":你传的 tools(如 web_search)、BASE_PYTHON_TOOLS(内置的 final_answer 就在这里,本质是个抛特殊异常/标记完成的 Python 函数)、你的 additional_args 变量(作为全局变量直接用)。state 字典跨步持久——上一步定义的变量下一步还在,这是"变量复用"优势(下节论证)的机制基础。
final_answer(...) 的调用会被 evaluate_python_code 捕获并把 is_final_answer 置真;fix_final_answer_code(local_python_executor.py:332-360)则处理一个经典坑:LLM 有时会写 final_answer = "xxx"(赋值)而不是 final_answer("xxx")(调用),正则把赋值与裸变量改名成 final_answer_variable、保留函数调用不动,避免污染工具名。
五、完整链路图
与 ToolCallingAgent 唯一的结构差异就在 ③④ 两格:动作从"JSON → 查表调度函数"变成"代码 → 解释器执行"。循环、记忆、错误处理全部复用第 2 章的地基——这是 MultiStepAgent 抽象的胜利。
💡 阶梯要点:CodeAgent 的全部魔法可以浓缩成一句:把工具调用从"协议"降维成"函数调用"。LLM 不再需要理解 JSON 工具协议,只需要会写 Python——调用参数是变量、返回值是对象、组合靠语法。代价是必须引入一个"能安全执行 LLM 代码"的执行器,这就是第 6 章 AST 解释器存在的理由。
本节要点回顾
- CodeAgent 只覆写单步语义;新增参数核心是
executor_type(默认 local)、additional_authorized_imports、code_block_tags(默认<code>)。 - 默认 import 白名单
BASE_BUILTIN_MODULES纯计算库 11 个;远程执行器下禁用 managed_agents。 - 单步四段:生成(闭标签当停止序列)→ 解析(
parse_code_blobs)→ 执行(python_executor(code))→ 观察(print 日志 + 末表达式截断)。 parse_code_blobs四级降级:配置标签 → markdown 围栏 → ast.parse 整段 → 带示例的报错。send_tools在run()时把 tools 与 managed_agents 注入解释器;state跨步持久,变量可复用。final_answer(...)终止循环;fix_final_answer_code修"赋值冒充调用"的坑。- 执行崩溃时 print 日志仍抢救进观察,LLM 可据其精准改错重试。
- 本地执行器不是安全沙箱,生产用四种远程 executor(第 6 章)。
下一节回答全书最核心的"为什么":同样跑一个任务,代码 action 为什么比 JSON 工具调用少约 30% 步?四个本质优势逐一拆解。