上一篇我们看完了 cli.py，知道了 Agent 是怎么被组装起来的。这一篇我们进入 agents.py，看看 Agent 到底是怎么”跑”起来的——这里才是 smolagents 的核心。

这个方法是 agents.py 的入口方法，从这里可以开始理解这个 agent 做了什么。

首先，这里对参数进行了一些预处理：

        max_steps = max_steps or self.max_steps
        self.task = task
        self.interrupt_switch = False
        if additional_args:
            self.state.update(additional_args)
            self.task += f"""
You have been provided with these additional arguments, that you can access directly using the keys as variables:
{str(additional_args)}."""

可以看到，这里把一些参数放到了 self.state 中，这样 LLM 就能够知道有哪些参数可以被使用，并且能够知道这些参数的类型和数据。以下我来举个例子：

import pandas as pd

df = pd.read_csv("sales.csv")
agent.run("找出销售额最高的三个月", additional_args={"df": df})

LLM 收到的任务是：

找出销售额最高的三个月
You have been provided with these additional arguments, that you can access directly using the keys as variables:
{'df':      month  sales
0    2024-01  15000
...}.

它就会写出：

top3 = df.nlargest(3, "sales")
final_answer(top3)

可以看到，变量名会原样出现在提示词里，LLM 靠名字猜用途，所以开发者提醒我们，变量名一定要按照规范去写，这样 LLM 才能猜得出这个变量的意图。并且这里会把参数所有的内容都给 LLM ，所以一个变量要是很大，他提交给 LLM 也会消耗很多 token 。

接下来就是初始化记忆的部分，smolagents 在这里把初始提示词加入到了记忆中：

        self.memory.system_prompt = SystemPromptStep(system_prompt=self.system_prompt)

我们看一下 self.system_prompt 方法的实现：

    @property
    def system_prompt(self) -> str:
        return self.initialize_system_prompt()

    @abstractmethod
    def initialize_system_prompt(self) -> str:
        """To be implemented in child classes"""
        ...

class ToolCallingAgent(MultiStepAgent):
    def initialize_system_prompt(self) -> str:
        system_prompt = populate_template(
            self.prompt_templates["system_prompt"],
            variables={
                "tools": self.tools,
                "managed_agents": self.managed_agents,
                "custom_instructions": self.instructions,
            },
        )
        return system_prompt

class CodeAgent(MultiStepAgent):
    def initialize_system_prompt(self) -> str:
        system_prompt = populate_template(
            self.prompt_templates["system_prompt"],
            variables={
                "tools": self.tools,
                "managed_agents": self.managed_agents,
                "authorized_imports": (
                    "You can import from any package you want."
                    if "*" in self.authorized_imports
                    else str(self.authorized_imports)
                ),
                "custom_instructions": self.instructions,
                "code_block_opening_tag": self.code_block_tags[0],
                "code_block_closing_tag": self.code_block_tags[1],
            },
        )
        return system_prompt

可以看到，对于 ToolCallingAgent 和 CodeAgent ，分别实现了两套组装系统提示词的方法。

我们再看一下 SystemPromptStep() 方法的实现：

@dataclass
class SystemPromptStep(MemoryStep):
    system_prompt: str

    def to_messages(self, summary_mode: bool = False) -> list[ChatMessage]:
        if summary_mode:
            return []
        return [ChatMessage(role=MessageRole.SYSTEM, content=[{"type": "text", "text": self.system_prompt}])]

所以运行完 self.memory.system_prompt = SystemPromptStep(system_prompt=self.system_prompt) 之后，self.memory.system_prompt 是这样：

SystemPromptStep(system_prompt="You are an expert assistant who can solve any task using code blobs...")

有一堆提示词来教 Agent 如何工作，这部分提示词写在了 prompts 文件夹下的 yaml 文件中。可以观察到，这里没有把用户本次的任务加进去，换句话说，系统提示词不含任务，同一个 agent 跑任何任务，系统提示词一模一样。因为这里有分工：

system_prompt	“你是谁、你有什么工具、你该怎么工作”	不变
TaskStep	“这次要干什么”	每次都变

如果 reset ，则需要清空对话历史和统计数据：

        if reset:
            self.memory.reset()
            self.monitor.reset()

我们看一下这两个方法的实现：

class AgentMemory:
    def reset(self):
        """Reset the agent's memory, clearing all steps and keeping the system prompt."""
        self.steps = []

class Monitor:
    def reset(self):
        self.step_durations = []
        self.total_input_token_count = 0
        self.total_output_token_count = 0

可以看到，注释特意说了，不清空系统提示词。因为系统提示词根本不在 steps 里面，它是 AgentMemory 上一个独立的属性。所以 self.steps = [] 不会清除系统提示词，我们可以看一下 AgentMemory 的定义：

class AgentMemory:
    def __init__(self, system_prompt: str):
        self.system_prompt: SystemPromptStep = SystemPromptStep(system_prompt=system_prompt)
        self.steps: list[TaskStep | ActionStep | PlanningStep] = []

清空统计数据非常好理解，将每步耗时、输入 token 总数、输出 token 总数都清空了。

接下来的代码纯粹是一段打印日志：

        self.logger.log_task(
            content=self.task.strip(),
            subtitle=f"{type(self.model).__name__} - {(self.model.model_id if hasattr(self.model, 'model_id') else '')}",
            level=LogLevel.INFO,
            title=self.name if hasattr(self, "name") else None,
        )

刚才说过，任务没有被加入到系统提示词中，因为任务可能因为下次对话而发生变化，因此这里单独把任务放在记忆中：

        self.memory.steps.append(TaskStep(task=self.task, task_images=images))

这里 TaskStep 就被放在 memory.steps 里了，我们看一下 TaskStep 类的实现：

@dataclass
class TaskStep(MemoryStep):
    task: str
    task_images: list["PIL.Image.Image"] | None = None

    def to_messages(self, summary_mode: bool = False) -> list[ChatMessage]:
        content = [{"type": "text", "text": f"New task:\n{self.task}"}]
        if self.task_images:
            content.extend([{"type": "image", "image": image} for image in self.task_images])

        return [ChatMessage(role=MessageRole.USER, content=content)]

可以看到，和 SystemPromptStep 类一样，都提供了一个 to_messages 方法。

LLM 有写python代码的能力，但是如何运行python代码呢？有两个思路，第一种比较简单，通过 docker 等虚拟化方式，在沙盒中创建一个.py文件，然后运行。第二种比较麻烦，也是 smolagents 默认的方法，就是写一个解释器，去处理 LLM 写的代码，打个比方：

LLM 写了以下代码： x = x + y ，我们写个解释器，根据数据结构的方法，把这个代码的运算拆分出来，也就是x + y 和 x = 两个运算，那么我们写一个方法来处理x + y 运算，再写一个方法来处理x = 运算，就可以做到这个运算了。

smolagents 默认的解释器在 local_python_executor.py 中，他并没有完全实现所有的 python 语法，但是实现了常用的，够用就行了。我们再回来看看填充解释器的代码：

        if getattr(self, "python_executor", None):
            self.python_executor.send_variables(variables=self.state)
            self.python_executor.send_tools({**self.tools, **self.managed_agents})

send_variables() 方法把变量放进字典， send_tools() 方法把工具放进字典。这样 CodeAgent 就能调用了。

如果设定的是流式输出，那么现在就可以直接进入主循环了，也就是代码中的 _run_stream() 方法：

        if stream:
            # The steps are returned as they are executed through a generator to iterate on.
            return self._run_stream(task=self.task, max_steps=max_steps, images=images)

我们现在先来看看，如果不执行流式输出之后还需要做什么，待会儿我们再来看 _run_stream() 方法，因为其实无论是流式输出还是非流式输出，都是需要依靠这个方法来继续的。而非流式输出只是帮我们在后台静默地运行完所有的步骤之后，展示给我们。

这个没什么好说的，记录一下开始的当前时间，待会儿通过结束时间可以算出一共执行了多久。

        run_start_time = time.time()

        steps = list(self._run_stream(task=self.task, max_steps=max_steps, images=images))

这个是非流式输出的核心代码，list() 拿到一个生成器，会从头到尾迭代一遍，把每次返回出来的东西装进列表。等价于：

steps = []
for item in self._run_stream(...):     # 每转一圈，agent 走一步
    steps.append(item)

steps 的结构我们这里先给出一个示例：

[0] ToolCall          ┐
[1] ActionOutput      ├─ 第 1 步的三个事件
[2] ActionStep        ┘
[3] ToolCall          ┐
[4] ActionOutput      ├─ 第 2 步的三个事件
[5] ActionStep        ┘
[6] FinalAnswerStep   ← 收尾，最后一个

可以看到最后一个 step 的类型是 FinalAnswerStep ，这就是模型最终的输出结果。smolagents 接下来就要对这个结果进行处理，所以先在这里做了一个判断，如果类型不符，直接报错：

        # Outputs are returned only at the end. We only look at the last step.
        assert isinstance(steps[-1], FinalAnswerStep)
        output = steps[-1].output

那个地方用了一个很简洁的编程方式，但是我觉得可读性不是很高：

        return_full_result = return_full_result if return_full_result is not None else self.return_full_result

等价于：

if return_full_result is not None:
    return_full_result = return_full_result      # 用参数传的
else:
    return_full_result = self.return_full_result # 用构造时的设置

这里一大段代码都是在统计 token 的消耗：

        if return_full_result:
            total_input_tokens = 0
            total_output_tokens = 0
            correct_token_usage = True
            for step in self.memory.steps:
                if isinstance(step, (ActionStep, PlanningStep)):
                    if step.token_usage is None:
                        correct_token_usage = False
                        break
                    else:
                        total_input_tokens += step.token_usage.input_tokens
                        total_output_tokens += step.token_usage.output_tokens
            if correct_token_usage:
                token_usage = TokenUsage(input_tokens=total_input_tokens, output_tokens=total_output_tokens)
            else:
                token_usage = None

可以看到，只有 ActionStep 和 PlanningStep需要统计 token 的消耗，因为只有这两种 step 是调用了 LLM 的。像 TaskStep 这种 step ，他没有调用 LLM ，就自然不需要去统计token的消耗。

另外，我们还可以观察到，这里有一个 correct_token_usage 标志位，这里的设计是防止某一次 step 中 LLM 没有返回 token 消耗，这代表着有可能是出现了一些问题，所以它这里提供了一个标志位，用于记录是否出现了出现问题的情况，以防止问题被这样掩盖下去了。

接下来 smolagents 还判断了一下终止原因，判断这次运行是”正常答完”还是”跑满步数被迫收场”：

            if self.memory.steps and isinstance(getattr(self.memory.steps[-1], "error", None), AgentMaxStepsError):
                state = "max_steps_error"
            else:
                state = "success"

smolagents 把 memory 里的 Step 对象，全部转换成普通的字典，方便用户未来转为 Json 格式：

            step_dicts = self.memory.get_full_steps()

class AgentMemory:
    def get_full_steps(self) -> list[dict]:
        """Return a full representation of the agent's steps, including model input messages."""
        if len(self.steps) == 0:
            return []
        return [step.dict() for step in self.steps]

如果是 return_full_result ，那么就返回 RunResult 对象：

            return RunResult(
                output=output,
                token_usage=token_usage,
                steps=step_dicts,
                timing=Timing(start_time=run_start_time, end_time=time.time()),
                state=state,
            )

刚刚我们看了 run() 方法，可以发现它事实上把整个运行的步骤都丢给了 _run_stream() 方法，那么我们现在就来看看这个 _run_stream() 方法，到底是如何让整个Agent循环跑起来的。

在第一次循环开始前，需要先将步数设为1，并且将 returned_final_answer 设为 False。

self.step_number = 1
returned_final_answer = False

接下来就是开始循环了，每次循环前需要先检查是否需要中断。如果 returned_final_answer 为 True 或者执行的步数超过配置的最大步数的时候，就不继续进行循环了：

        while not returned_final_answer and self.step_number <= max_steps:

由于这个循环开始之后，我们可能还希望有一个方法能够中断。所以 smolagents 在这里准备了一个 self.interrupt_switch ，如果我们不想让它继续循环了，就可以调用 interrupt(self) 方法，这样他就不会继续循环了。但是我看了代码，并没有任何一个地方调用了这个方法，所以这里事实上是留给二次开发的一个锚，目前这个项目并没有用到。

            if self.interrupt_switch:
                raise AgentError("Agent interrupted.", self.logger)

接下来就判断当前循环需不需要进行规划 step ，如果 planning_interval=3 ，那么就在第 1、4、7 步规划。

这个地方我研究了很久，因为 _generate_planning_step() 这个方法确实是有点大，它针对2×2=4种情况都做了处理，但是其实我们看它的返回值还是很清晰的：

    def _generate_planning_step(
        self, task, is_first_step: bool, step: int
    ) -> Generator[ChatMessageStreamDelta | PlanningStep]:
        start_time = time.time()
        if is_first_step:
            input_messages = [
                ChatMessage(
                    role=MessageRole.USER,
                    content=[
                        {
                            "type": "text",
                            "text": populate_template(
                                self.prompt_templates["planning"]["initial_plan"],
                                variables={"task": task, "tools": self.tools, "managed_agents": self.managed_agents},
                            ),
                        }
                    ],
                )
            ]
            if self.stream_outputs and hasattr(self.model, "generate_stream"):
                plan_message_content = ""
                output_stream = self.model.generate_stream(input_messages, stop_sequences=["<end_plan>"])  # type: ignore
                input_tokens, output_tokens = 0, 0
                with Live("", console=self.logger.console, vertical_overflow="visible") as live:
                    for event in output_stream:
                        if event.content is not None:
                            plan_message_content += event.content
                            live.update(Markdown(plan_message_content))
                            if event.token_usage:
                                input_tokens = event.token_usage.input_tokens
                                output_tokens += event.token_usage.output_tokens
                        yield event
            else:
                plan_message = self.model.generate(input_messages, stop_sequences=["<end_plan>"])
                plan_message_content = plan_message.content
                input_tokens, output_tokens = 0, 0
                if plan_message.token_usage:
                    input_tokens = plan_message.token_usage.input_tokens
                    output_tokens = plan_message.token_usage.output_tokens
            plan = textwrap.dedent(
                f"""Here are the facts I know and the plan of action that I will follow to solve the task:\n```\n{plan_message_content}\n```"""
            )
        else:
            # Summary mode removes the system prompt and previous planning messages output by the model.
            # Removing previous planning messages avoids influencing too much the new plan.
            memory_messages = self.write_memory_to_messages(summary_mode=True)
            plan_update_pre = ChatMessage(
                role=MessageRole.SYSTEM,
                content=[
                    {
                        "type": "text",
                        "text": populate_template(
                            self.prompt_templates["planning"]["update_plan_pre_messages"], variables={"task": task}
                        ),
                    }
                ],
            )
            plan_update_post = ChatMessage(
                role=MessageRole.USER,
                content=[
                    {
                        "type": "text",
                        "text": populate_template(
                            self.prompt_templates["planning"]["update_plan_post_messages"],
                            variables={
                                "task": task,
                                "tools": self.tools,
                                "managed_agents": self.managed_agents,
                                "remaining_steps": (self.max_steps - step),
                            },
                        ),
                    }
                ],
            )
            input_messages = [plan_update_pre] + memory_messages + [plan_update_post]
            if self.stream_outputs and hasattr(self.model, "generate_stream"):
                plan_message_content = ""
                input_tokens, output_tokens = 0, 0
                with Live("", console=self.logger.console, vertical_overflow="visible") as live:
                    for event in self.model.generate_stream(
                        input_messages,
                        stop_sequences=["<end_plan>"],
                    ):  # type: ignore
                        if event.content is not None:
                            plan_message_content += event.content
                            live.update(Markdown(plan_message_content))
                            if event.token_usage:
                                input_tokens = event.token_usage.input_tokens
                                output_tokens += event.token_usage.output_tokens
                        yield event
            else:
                plan_message = self.model.generate(input_messages, stop_sequences=["<end_plan>"])
                plan_message_content = plan_message.content
                input_tokens, output_tokens = 0, 0
                if plan_message.token_usage:
                    input_tokens = plan_message.token_usage.input_tokens
                    output_tokens = plan_message.token_usage.output_tokens
            plan = textwrap.dedent(
                f"""I still need to solve the task I was given:\n```\n{self.task}\n```\n\nHere are the facts I know and my new/updated plan of action to solve the task:\n```\n{plan_message_content}\n```"""
            )
        log_headline = "Initial plan" if is_first_step else "Updated plan"
        self.logger.log(Rule(f"[bold]{log_headline}", style="orange"), Text(plan), level=LogLevel.INFO)
        yield PlanningStep(
            model_input_messages=input_messages,
            plan=plan,
            model_output_message=ChatMessage(role=MessageRole.ASSISTANT, content=plan_message_content),
            token_usage=TokenUsage(input_tokens=input_tokens, output_tokens=output_tokens),
            timing=Timing(start_time=start_time, end_time=time.time()),
        )

他的返回值就是：Generator[ChatMessageStreamDelta | PlanningStep] 。所以如果它返回的是 ChatMessageStreamDelta ，那就说明它还在流式输出中，所以代码里面就直接用 yield 返回给了上级，当它最后运行完了，那么它返回的就是 PlanningStep ，那这就是我们想得到的规划步骤了，所以，他这里的代码是这么写的：

            # Run a planning step if scheduled
            if self.planning_interval is not None and (
                self.step_number == 1 or (self.step_number - 1) % self.planning_interval == 0
            ):
                planning_start_time = time.time()
                planning_step = None
                for element in self._generate_planning_step(
                    task, is_first_step=len(self.memory.steps) == 1, step=self.step_number
                ):  # Don't use the attribute step_number here, because there can be steps from previous runs
                    yield element
                    planning_step = element
                assert isinstance(planning_step, PlanningStep)

直到这里，我们就可以确认 planning_step 这个变量拿到的，一定是我们想得到的规划步骤类了。

最后再走一下统一的步骤收尾：

                planning_end_time = time.time()
                planning_step.timing = Timing(
                    start_time=planning_start_time,
                    end_time=planning_end_time,
                )
                self._finalize_step(planning_step)
                self.memory.steps.append(planning_step)

接下来就是执行 action step 了，先创建一个空的 ActionStep ，然后给 Agent 的 _step_stream 方法去执行，向上返回输出的每一个字符，直到输出的类是 ActionOutput ，说明模型已经完成了最终的输出。

# Start action step!
            action_step_start_time = time.time()
            action_step = ActionStep(
                step_number=self.step_number,
                timing=Timing(start_time=action_step_start_time),
                observations_images=images,
            )
            self.logger.log_rule(f"Step {self.step_number}", level=LogLevel.INFO)
            try:
                for output in self._step_stream(action_step):
                    # Yield all
                    yield output

                    if isinstance(output, ActionOutput) and output.is_final_answer:
                        final_answer = output.output
                        self.logger.log(
                            Text(f"Final answer: {final_answer}", style=f"bold {YELLOW_HEX}"),
                            level=LogLevel.INFO,
                        )

                        if self.final_answer_checks:
                            self._validate_final_answer(final_answer)
                        returned_final_answer = True
                        action_step.is_final_answer = True

smolagents 在这里处理了一些可能的错误情况，比如模型调用错误会直接往上抛，而调用工具错误这类型的错误则不会向上抛，而会记录到记忆中。最后还一个 finally 来收尾：

            except AgentGenerationError as e:
                # Agent generation errors are not caused by a Model error but an implementation error: so we should raise them and exit.
                raise e
            except AgentError as e:
                # Other AgentError types are caused by the Model, so we should log them and iterate.
                action_step.error = e
            finally:
                self._finalize_step(action_step)
                self.memory.steps.append(action_step)
                yield action_step
                self.step_number += 1

smolagents 在这里新增了一个检测，当他发现模型仍然没有做出最后的回答，但是步数已经达到所规定的上限的时候，它会强制要求模型作答，并且生成一个带错误提示的 ActionStep ：

        if not returned_final_answer and self.step_number == max_steps + 1:
            final_answer = self._handle_max_steps_reached(task)
            yield action_step

最后一步就是统一包装最终输出了：

        final_answer_step = FinalAnswerStep(handle_agent_output_types(final_answer))
        self._finalize_step(final_answer_step)
        yield final_answer_step

看了这么多，是时候自己尝试做一个小 Agent 了，这样能帮助我们更好地理解 Agent 开发。