全面解析smolagents库:MultiStepAgent、CodeAgent和ToolCallingAgent

2025年02月11日 由 alex 发表 4009 0

在人工智能领域,智能体(agents)是指与环境进行交互以实现特定目标的实体。HuggingFace的smolagents库为构建此类智能体提供了一个强大的框架。它提供了三种主要的智能体类型:多步智能体(MultiStepAgent)、代码智能体(CodeAgent)和工具调用智能体(ToolCallingAgent)。每种智能体都旨在以独特的方式解决任务,利用语言模型和工具来执行动作并解释结果。


14


这篇文章将探讨每种智能体的功能、实现方式以及用例。我们将深入代码,突出关键部分,并解释它们如何对智能体的整体行为做出贡献。此外,我们还将讨论在何种情况下应选择使用某种智能体而非其他智能体,以及它们之间的区别。


以下是它在简单 GIF 中的工作原理:


15


理解MultiStepAgent

MultiStepAgent是smolagents库中所有智能体的基础类。它实现了ReAct框架,该框架包括循环执行动作和观察,直到达成目标或达到最大步数。这个智能体为更专业的智能体(如CodeAgent和ToolCallingAgent)提供了蓝图。


关键特性

  1. ReAct框架:智能体根据语言模型的输入执行一系列动作,并观察环境的响应。
  2. 可定制工具:开发人员可以定义和集成智能体在执行过程中可以使用的自定义工具。
  3. 规划能力:智能体使用规划间隔定期规划其下一步动作,确保它保持在实现目标的轨道上。


代码亮点

为了理解当用户要求智能体解决任务时会发生什么,让我们看看输入到输出的流程是如何工作的。MultiStepAgent类旨在使用ReAct框架逐步解决任务,该框架包括思考(由语言模型生成的动作)和观察(从环境或工具获得的结果)的循环。下面,我将解释run()方法的工作流程,将其分解为主要步骤,包括规划、工具调用和其他重要部分。


run()中的初始化

MultiStepAgent类中的run方法是协调智能体执行任务整个过程的关键函数。它是运行任务的入口点,并管理操作流程,包括初始化、内存管理、日志记录和逐步执行。下面,我将详细解释该方法,将其分解为关键组件,并解释每个部分如何对整体功能做出贡献。


方法的目的

run方法负责使用ReAct框架执行给定任务。它初始化必要的组件,管理对话状态,并在完成所有步骤后,要么流式传输中间结果,要么返回最终结果。该方法确保智能体高效运行,并在执行过程中动态适应新信息。


关键组件


输入参数

该方法接受几个参数,这些参数定义了如何执行任务:

  • task(str):智能体需要解决的主要任务。这是用户提供的主要输入。
  • stream(bool,默认False):确定智能体是应该生成中间结果时即返回(流式模式),还是仅在结束时返回最终结果。
  • reset(bool,默认True):指定智能体是应该重置其内存并重新开始,还是从上一次运行的位置继续。
  • images(list[str],可选):可能与任务相关的图像路径。
  • additional_args(dict,可选):智能体在执行过程中可以使用的任何额外变量(例如,数据框、图像)。这些被添加到智能体的状态字典中。


任务初始化

方法首先将任务分配给智能体的self.task属性。如果提供了additional_args,则将它们添加到智能体的状态字典(self.state)中,并附加到任务描述中。这确保了智能体在执行过程中可以访问这些变量。


if additional_args is not None:
    self.state.update(additional_args)
    self.task += f"""
You have been provided with these additional arguments, that you can access using the keys as variables in your python code:
{str(additional_args)}."""


这一步对于需要外部输入的任务(如数据分析或图像处理)特别有用。


系统提示初始化

系统提示是智能体操作的关键组成部分。它为语言模型(LLM)提供上下文和指令,指导如何执行任务。initialize_system_prompt方法根据预定义的模板和智能体的配置生成系统提示。


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


系统提示作为SystemPromptStep存储在智能体的内存中,确保以后可以随时参考。


内存和监控重置

如果reset参数设置为True,智能体会清空其内存并重置监控指标。这确保了智能体在执行新任务时能够从一个全新的状态开始。


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


当任务与之前的运行无关时,重置内存是至关重要的,而重置监控则能确保性能指标的准确跟踪。


记录任务详情

智能体会记录任务的详细信息,以提高透明度和便于调试。这包括任务内容、所使用的模型,以及任何可选的标题或副标题。


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 '')}",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,
)


将任务添加到内存中

任务作为TaskStep添加到智能体的内存中。这一步包括任何相关的图像,并作为智能体执行的起点。


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


将任务存储在内存中确保其在规划和执行过程中可以随时参考。


执行模式

该方法支持两种执行模式:流式和非流式。


a. 流式模式


如果stream=True,该方法会调用_run生成器函数,该函数在生成中间结果时即时返回。这允许用户实时观察智能体的进度。


if stream:
    return self._run(task=self.task, images=images)


流式模式对于需要持续反馈或监控的任务非常有用。


b. 非流式模式


如果stream=False,该方法会收集所有中间结果,并仅返回最终输出。这是通过使用双端队列(deque)来提取最后一步的结果实现的。


return deque(self._run(task=self.task, images=images), maxlen=1)[0]


非流式模式适用于只需要最终结果的场景。


与_run的交互


_run方法是一个生成器函数,负责实际任务的逐步执行。它处理以下内容:

  1. 规划步骤:根据planning_interval定期更新计划。
  2. 动作生成:使用语言模型生成动作。
  3. 工具执行:根据生成的动作执行工具。
  4. 观察记录:记录观察和错误,以供将来参考。
  5. 提供最终答案:如果达到最大步数仍未解决任务,则提供最终答案。


run方法将核心执行逻辑委托给_run,确保模块化和关注点分离。我们将在后文中深入探讨_run方法。


规划步骤

在进入主要执行循环之前,如果配置为这样做(设置了planning_interval),智能体可能会执行一个初始规划步骤。MultiStepAgent类中的planning_step方法是智能体决策过程的关键组成部分。它负责定期生成或更新计划,以引导智能体解决给定任务。该方法确保智能体以结构化和目标导向的方式运作,利用其内存和语言模型(LLM)来制定计划,并根据需要进行更新。


让我们详细分解该方法:


方法的目的

planning_step方法在智能体执行期间按planning_interval参数确定的定期调用。其主要目的是:

  1. 生成初始计划:在第一步,智能体根据任务、可用工具和任何初始已知事实创建高级计划。
  2. 更新计划:在后续步骤中,智能体通过纳入新观察、事实和迄今为止的进展来修订其计划。
  3. 记录计划和事实:该方法记录生成的计划和事实,以提高透明度和便于调试。


这种结构化方法有助于智能体专注于任务,并动态适应新信息。


关键组件


输入

该方法接受三个参数:

  • task:描述智能体需要解决的任务的字符串。
  • is_first_step:布尔值,指示这是否是第一个规划步骤。如果为True,则方法生成初始计划;否则,更新现有计划。
  • step:当前步骤编号,用于计算剩余步骤并为规划提供上下文。


初始规划

如果is_first_step为True,方法执行以下步骤:


a. 提取初始事实

智能体首先使用语言模型提取有关任务的初始事实:


message_prompt_facts = {
    "role": MessageRole.SYSTEM,"role": MessageRole.SYSTEM,
    "content": [{"type": "text", "text": self.prompt_templates["planning"]["initial_facts"]}],
}
input_messages = [message_prompt_facts]
chat_message_facts: ChatMessage = self.model(input_messages)
answer_facts = chat_message_facts.content


  • message_prompt_facts模板提示语言模型(LLM)生成与任务相关的事实。
  • 语言模型处理提示并返回提取的事实(answer_facts)。


b. 生成初始计划

智能体使用提取的事实来制定初始计划:


message_prompt_plan = {
    "role": MessageRole.USER,"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,
                    "answer_facts": answer_facts,
                },
            ),
        }
    ],
}
chat_message_plan: ChatMessage = self.model([message_prompt_plan], stop_sequences=["<end_plan>"])
answer_plan = chat_message_plan.content


  • message_prompt_plan模板结合任务、工具、管理的智能体和提取的事实来创建一个详细的计划。
  • 语言模型处理提示并生成计划(answer_plan)。


c. 记录和存储计划

生成的计划和事实被记录和存储在智能体的内存中:


final_plan_redaction = f"""Here is the plan of action that I will follow to solve the task:f"""Here is the plan of action that I will follow to solve the task:
```
{answer_plan}
```"""
final_facts_redaction = f"""Here are the facts that I know so far:
```
{answer_facts}
```""".strip()
self.memory.steps.append(
    PlanningStep(
        model_input_messages=input_messages,
        plan=final_plan_redaction,
        facts=final_facts_redaction,
        model_output_message_plan=chat_message_plan,
        model_output_message_facts=chat_message_facts,
    )
)
self.logger.log(
    Rule("[bold]Initial plan", style="orange"),
    Text(final_plan_redaction),
    level=LogLevel.INFO,
)


  • 计划和事实被格式化为可读的字符串,并作为PlanningStep添加到智能体的内存中。
  • 日志记录器记录计划以提高可见性。


更新计划

如果is_first_step为False,该方法通过纳入新信息来更新现有计划:


a. 提取更新后的事实

智能体从其内存中检索更新后的事实,并将它们与预定义的提示相结合:


facts_update_pre_messages = {
    "role": MessageRole.SYSTEM,"role": MessageRole.SYSTEM,
    "content": [{"type": "text", "text": self.prompt_templates["planning"]["update_facts_pre_messages"]}],
}
facts_update_post_messages = {
    "role": MessageRole.SYSTEM,
    "content": [{"type": "text", "text": self.prompt_templates["planning"]["update_facts_post_messages"]}],
}
input_messages = [facts_update_pre_messages] + memory_messages + [facts_update_post_messages]
chat_message_facts: ChatMessage = self.model(input_messages)
facts_update = chat_message_facts.content


  • 内存消息包括智能体过去的动作、观察和错误。
  • 语言模型处理组合输入以生成更新后的事实(facts_update)。


b. 生成更新后的计划

智能体使用更新后的事实来修订其计划:


update_plan_pre_messages = {
    "role": MessageRole.SYSTEM,"role": MessageRole.SYSTEM,
    "content": [
        {
            "type": "text",
            "text": populate_template(
                self.prompt_templates["planning"]["update_plan_pre_messages"], variables={"task": task}
            ),
        }
    ],
}
update_plan_post_messages = {
    "role": MessageRole.SYSTEM,
    "content": [
        {
            "type": "text",
            "text": populate_template(
                self.prompt_templates["planning"]["update_plan_pre_messages"],
                variables={
                    "task": task,
                    "tools": self.tools,
                    "managed_agents": self.managed_agents,
                    "facts_update": facts_update,
                    "remaining_steps": (self.max_steps - step),
                },
            ),
        }
    ],
}
chat_message_plan: ChatMessage = self.model(
    [update_plan_pre_messages] + memory_messages + [update_plan_post_messages],
    stop_sequences=["<end_plan>"],
)
answer_plan = chat_message_plan.content


  • 更新后的计划会考虑剩余步骤(self.max_steps - step)并融入最新的事实。
  • 语言模型处理提示并生成修订后的计划(answer_plan)。


c. 记录和存储更新后的计划

更新后的计划和事实被记录和存储在智能体的内存中:


final_plan_redaction = textwrap.dedent(
    f"""I still need to solve the task I was given:f"""I still need to solve the task I was given:
    ```
    {task}
    ```
    Here is my new/updated plan of action to solve the task:
    ```
    {answer_plan}
    ```"""
)
final_facts_redaction = textwrap.dedent(
    f"""Here is the updated list of the facts that I know:
    ```
    {facts_update}
    ```"""
)
self.memory.steps.append(
    PlanningStep(
        model_input_messages=input_messages,
        plan=final_plan_redaction,
        facts=final_facts_redaction,
        model_output_message_plan=chat_message_plan,
        model_output_message_facts=chat_message_facts,
    )
)
self.logger.log(
    Rule("[bold]Updated plan", style="orange"),
    Text(final_plan_redaction),
    level=LogLevel.INFO,
)


  • 更新后的计划和事实被格式化并添加到智能体的内存中。
  • 日志记录器记录更新后的计划以提高可见性。


核心执行逻辑

MultiStepAgent类中的_run方法是使用ReAct框架逐步执行任务的关键函数。它作为一个生成器运行,在每一步产生中间结果,并确保智能体在最大步数(max_steps)的限制下朝着解决任务的方向前进。下面,我将详细解释该方法,将其分解为关键组件,并解释每个部分如何贡献于整体功能。


方法的目的

_run方法负责以结构化和迭代的方式执行给定任务。它管理以下内容:

  1. 规划:根据智能体的进展定期更新计划。
  2. 动作执行:使用语言模型(LLM)生成动作并执行。
  3. 观察记录:记录观察结果、错误和其他相关信息以供将来参考。
  4. 提供最终答案:如果任务解决或达到最大步数仍未成功,则提供最终答案。


作为生成器运行,该方法允许用户实时观察智能体的进展(流式模式)或仅检索最终结果(非流式模式)。


关键组件


初始化

方法首先初始化将在整个执行过程中使用的变量:

  • final_answer:跟踪任务是否已解决。初始设置为None。
  • self.step_number:跟踪当前步数,从1开始。
  • memory_step:表示智能体记忆中的当前步骤,包括步数、开始时间和相关图像等详细信息。


final_answer = NoneNone
self.step_number = 1
while final_answer is None and self.step_number <= self.max_steps:
    step_start_time = time.time()
    memory_step = ActionStep(
        step_number=self.step_number,
        start_time=step_start_time,
        observations_images=images,
    )


这一初始化确保智能体每一步都从空白状态开始,并随时间跟踪进度。


规划步骤

如果定义了规划间隔(planning_interval),智能体会定期更新其计划。这会在当前步数是规划间隔的倍数时发生。


if self.planning_interval is not None and self.step_number % self.planning_interval == 1:
    self.planning_step(
        task,
        is_first_step=(self.step_number == 1),
        step=self.step_number,
    )


  • 初始规划:在第一步,智能体根据任务、可用工具和任何初始事实生成初始计划。
  • 更新规划:在后续步骤中,智能体通过纳入新观察、事实和迄今为止的进展来修订其计划。


规划步骤方法(之前已描述)确保智能体以结构化和目标导向的方式运作,并动态适应新信息。


记录步骤

智能体记录当前步数,以提高透明度和便于调试:


self.logger.log_rule(f"Step {self.step_number}", level=LogLevel.INFO)f"Step {self.step_number}", level=LogLevel.INFO)


记录提供了对智能体活动的可见性,并帮助开发人员理解其决策过程。


执行一步

_run方法的核心是执行一步,这包括生成一个动作并观察结果:


final_answer = self.step(memory_step)


在子类中实现(例如,后面将解释的CodeAgent、ToolCallingAgent)的step方法执行以下操作:

  1. 动作生成:使用语言模型(LLM)根据智能体内存的当前状态生成一个动作。
  2. 工具执行:使用适当的工具执行生成的动作。
  3. 观察记录:在智能体内存中记录动作的结果(观察)。


如果动作产生了最终答案,智能体会提前终止并返回结果。


最终答案验证

如果生成了最终答案,智能体会使用一组预定义的检查(final_answer_checks)来验证它:


if final_answer is not None and self.final_answer_checks is not None:
    for check_function in self.final_answer_checks:
        try:
            assert check_function(final_answer, self.memory)
        except Exception as e:
            final_answer = None
            raise AgentError(f"Check {check_function.__name__} failed with error: {e}", self.logger)


这些检查确保最终答案在被接受之前符合特定标准。如果任何检查失败,智能体会继续寻找解决方案。


错误处理

如果在执行过程中发生错误,智能体会将其记录在memory_step对象中:


except AgentError as e:
    memory_step.error = e


这确保了错误被记录,并且可以在以后进行审查以用于调试目的。


完成步骤

完成步骤后,智能体将memory_step对象定稿并添加到其内存中:


finally:
    memory_step.end_time = time.time()
    memory_step.duration = memory_step.end_time - step_start_time
    self.memory.steps.append(memory_step)


智能体还会触发任何已注册的回调函数,使外部系统能够动态地监控或修改其行为:


for callback in self.step_callbacks:
    if len(inspect.signature(callback).parameters) == 1:
        callback(memory_step)
    else:
        callback(memory_step, agent=self)


最后,智能体增加步数并产出已完成的memory_step:


self.step_number += 11
yield memory_step


处理最大步数

如果智能体在未完成任务的情况下达到了最大步数,它会尝试根据其内存提供一个最终答案:


if final_answer is None and self.step_number == self.max_steps + 1:
    error_message = "Reached max steps."
    final_answer = self.provide_final_answer(task, images)
    final_memory_step = ActionStep(
        step_number=self.step_number, error=AgentMaxStepsError(error_message, self.logger)
    )
    final_memory_step.action_output = final_answer
    final_memory_step.end_time = time.time()
    final_memory_step.duration = memory_step.end_time - step_start_time
    self.memory.steps.append(final_memory_step)
    for callback in self.step_callbacks:
        if len(inspect.signature(callback).parameters) == 1:
            callback(final_memory_step)
        else:
            callback(final_memory_step, agent=self)
    yield final_memory_step


provide_final_answer方法总结智能体的交互,并根据可用信息生成响应。


产出最终答案

最后,在处理完所有步骤后,该方法产出最终答案:


yield handle_agent_output_types(final_answer)


这确保了输出格式正确,随时可供调用者使用。


_run方法是MultiStepAgent功能的核心。通过管理任务执行、内存、日志记录和错误处理,它确保智能体高效运行,并动态适应新信息。此方法展示了ReAct框架如何结合推理、动作和观察来解决复杂问题,使其成为广泛应用于各种场景的强大工具。


其他方法

类中还有其他方法,我将不深入探讨,仅在高层次上进行解释:


initialize_system_prompt

此方法旨在由子类(例如,CodeAgent、ToolCallingAgent)实现。它生成系统提示,为语言模型(LLM)提供上下文和指令。系统提示通常包括有关任务、可用工具以及智能体的任何约束或指南的详细信息。


write_memory_to_messages

此方法将智能体的内存转换为适合作为LLM输入的格式。它包括过去的动作、观察、错误和计划,确保LLM可以访问智能体的历史记录。这有助于智能体保持连续性并避免冗余步骤。


visualize

visualize方法创建智能体结构的丰富树状可视化,包括其工具、管理的智能体和内存。这对于调试和理解智能体的内部状态特别有用。


extract_action

此方法解析LLM的输出以提取动作和理由。它使用预定义标记(例如,“Action:”)拆分输出,并确保动作格式正确以供执行。如果输出不符合预期格式,则会引发错误。


provide_final_answer

当智能体在未完成任务的情况下达到最大步数时,此方法尝试根据智能体的内存提供最终答案。它构建总结智能体交互的提示,并要求LLM生成响应。


execute_tool_call

此方法使用提供的参数执行工具调用。它将参数中的占位符替换为智能体状态中的实际值,并调用适当的工具或管理的智能体。如果工具调用失败,会记录详细的错误信息以帮助诊断问题。


replay

replay方法提供智能体动作、观察和错误的逐步重放。如果detailed参数设置为True,它还会显示每一步的智能体内存。这主要用于调试和分析。


call

此方法允许智能体作为另一个智能体管理的智能体被调用。它为被管理的智能体添加额外的提示,运行任务,并以标准化格式包装输出。这对于一个智能体管理其他智能体的分层智能体架构特别有用。


辅助方法

  • get_variable_names:从Jinja2模板字符串中提取变量名。
  • populate_template:使用提供的变量呈现Jinja2模板,确保动态生成提示。
  • handle_agent_output_types:在处理并返回给用户之前,处理智能体的输出以处理不同类型(例如,文本、图像、音频)。


ToolCallingAgent

ToolCallingAgent是smolagents库中的一个专门智能体,它利用类似JSON的工具调用来与外部工具和管理的智能体进行交互。它基于基础的MultiStepAgent类构建,并引入了使用底层语言模型(LLM)能力生成、执行和管理工具调用的特定机制。下面,我将详细解释ToolCallingAgent的关键组件和功能。


ToolCallingAgent的目的

ToolCallingAgent旨在通过以结构化方式调用预定义工具或管理的智能体来解决问题。它利用LLM生成类似JSON的工具调用的能力,确保其交互具有清晰性和精确性。这使其特别适用于智能体需要与API、数据库或其他外部系统进行交互的场景。


关键特性

1. 类似JSON的工具调用:

  • 智能体以类似JSON的格式生成工具调用,其中包括工具名称、参数和标识符(tool_call_id)。
  • 这种结构化方法确保智能体可以一致地调用工具并解释其输出。


2. 动态工具执行:

  • 智能体根据生成的工具调用动态执行工具。
  • 如果工具调用指定了“最终答案”,智能体会提前终止并返回结果。


3. 状态管理:

  • 智能体维护一个状态字典(self.state)以存储中间结果,如工具输出或观察。
  • 这允许智能体在生成后续动作时引用之前的结果。


4. 日志记录和透明度:

  • 智能体记录每一步,包括工具调用、观察和错误,提供其决策过程的可见性。


5. 回退机制:

  • 如果智能体在未完成任务的情况下达到最大步数,它会尝试根据其内存提供最终答案。


方法详细解释


initialize_system_prompt

此方法生成为LLM提供上下文和指令的系统提示。它用有关可用工具和管理智能体的详细信息填充提示模板:


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


  • populate_template 函数使用提供的变量渲染 Jinja2 模板,确保系统提示的动态生成。系统提示作为大型语言模型(LLM)的指导,帮助其理解任务、可用工具和约束条件。


step

step 方法执行 ReAct 框架的一次迭代,包括动作生成、工具执行和观察记录。以下是详细分解:


a. 生成模型输出

智能体通过将其内存转换为适合大型语言模型(LLM)的格式来准备输入信息。然后,它生成一个响应,其中包括下一个工具调用:


model_message: ChatMessage = self.model(
    memory_messages,
    tools_to_call_from=list(self.tools.values()),list(self.tools.values()),
    stop_sequences=["Observation:"],
)


  • tools_to_call_from 参数指定了大型语言模型(LLM)可以调用的可用工具列表。
  • stop_sequences 参数确保当 LLM 遇到“Observation:”标记时停止生成输出。


b. 解析工具调用

智能体从模型的输出中解析出工具调用:


tool_call = model_message.tool_calls[0]0]
tool_name, tool_call_id = tool_call.function.name, tool_call.id
tool_arguments = tool_call.function.arguments


  • 工具调用包括工具名称、参数和一个标识符(tool_call_id)。
  • 如果模型没有生成任何工具调用,则会引发异常。


c. 记录工具调用

智能体记录工具调用以提高透明度:


self.logger.log(
    Panel(Text(f"Calling tool: '{tool_name}' with arguments: {tool_arguments}")),f"Calling tool: '{tool_name}' with arguments: {tool_arguments}")),
    level=LogLevel.INFO,
)


该日志提供了对智能体行为的可见性,使得调试和分析其行为变得更加容易。


d. 处理最终答案

如果工具调用指定了“最终答案”,智能体将提取该答案并返回:


if tool_name == "final_answer":
    if isinstance(tool_arguments, dict):
        if "answer" in tool_arguments:
            answer = tool_arguments["answer"]
        else:
            answer = tool_arguments
    else:
        answer = tool_arguments
    if (
        isinstance(answer, str) and answer in self.state.keys()
    ):  # if the answer is a state variable, return the value
        final_answer = self.state[answer]
        self.logger.log(
            f"[bold {YELLOW_HEX}]Final answer:[/bold {YELLOW_HEX}] Extracting key '{answer}' from state to return value '{final_answer}'.",
            level=LogLevel.INFO,
        )
    else:
        final_answer = answer
        self.logger.log(
            Text(f"Final answer: {final_answer}", style=f"bold {YELLOW_HEX}"),
            level=LogLevel.INFO,
        )
    memory_step.action_output = final_answer
    return final_answer


  • 如果答案是状态变量,智能体将从状态字典中检索其值。
  • 最终答案会被记录并返回给调用者。


e. 执行其他工具

对于非最终工具调用,智能体执行指定的工具并处理结果:


observation = self.execute_tool_call(tool_name, tool_arguments)
observation_type = type(observation)type(observation)
if observation_type in [AgentImage, AgentAudio]:
    if observation_type == AgentImage:
        observation_name = "image.png"
    elif observation_type == AgentAudio:
        observation_name = "audio.mp3"
    self.state[observation_name] = observation
    updated_information = f"Stored '{observation_name}' in memory."
else:
    updated_information = str(observation).strip()
self.logger.log(
    f"Observations: {updated_information.replace('[', '|')}",
    level=LogLevel.INFO,
)
memory_step.observations = updated_information
return None


  • execute_tool_call 方法使用提供的参数调用工具。
  • 如果观察结果是图像或音频,则将其存储在状态字典中以便将来参考。
  • 观察结果被记录并添加到智能体的记忆中。


用例

— API 交互:

  • ToolCallingAgent 非常适合与 RESTful API 或其他网络服务进行交互。其结构化的工具调用确保了精确和可靠的通信。


— 任务自动化:

  • 智能体可以自动化涉及多个工具或服务的工作流,如数据处理、文件管理或报告生成。


— 复杂问题解决:

  • 对于需要结构化推理和与外部系统交互的任务,ToolCallingAgent 提供了一个稳健的解决方案。


ToolCallingAgent 是一个强大的工具,用于构建能够以结构化和精确的方式与外部系统进行交互的智能体。通过利用类似 JSON 的工具调用,它确保了操作的清晰性和可靠性,使其成为自动化工作流、与 API 交互和解决复杂问题的绝佳选择。其与 ReAct 框架的集成以及全面的日志记录进一步增强了其功能,使开发人员能够构建高效且有效的解决方案。


CodeAgent

CodeAgent 是 smolagents 库中的一个专门智能体,旨在通过生成和执行 Python 代码片段来解决问题。与依赖类似 JSON 的工具调用的 ToolCallingAgent 不同,CodeAgent 专注于解析和执行由语言模型(LLM)生成的代码。这使得它特别适合计算任务,如数学计算、数据分析或模拟。该智能体基于一篇名为“可执行代码动作激发更好的大型语言模型智能体”的论文。


15


以下,我将详细解释 CodeAgent 的关键组件和功能,重点介绍其独特特性以及它如何在 ReAct 框架中运作。


CodeAgent 的目的

CodeAgent 专为需要计算推理或自动化的任务而设计,如数学计算、数据分析或模拟。它利用大型语言模型(LLM)生成 Python 代码的能力,并使用本地解释器或远程执行器来执行这些代码。这使得它特别适合需要精确和动态计算的场景。


关键特性

— 代码生成:

  • 智能体根据任务描述和可用工具生成 Python 代码片段。
  • 代码被动态解析和执行,实现实时问题解决。


— 授权导入:

  • 智能体支持一列授权导入(additional_authorized_imports),以确保代码的安全和受控执行。
  • 如果所有导入都被授权(“*”),智能体可以导入任何包,但如果环境中未安装该包,可能会导致问题。


— 远程执行:

  • 智能体可选择使用 E2BExecutor 进行远程代码执行,这对于扩展或在隔离环境中运行代码非常有用。


— 状态管理:

  • 智能体维护一个状态字典(self.state)来存储中间结果,如执行代码片段的输出。
  • 这允许智能体在生成后续动作时引用之前的结果。


— 日志记录和透:

  • 智能体记录每一步,包括生成的代码、执行日志和错误,提供其决策过程的可见性。


— 回退机制:

  • 如果智能体在未完成任务的情况下达到最大步数,它会尝试根据其记忆提供一个最终答案。


16


方法详解:


初始化(init)

构造函数用必要的组件和配置初始化智能体:


— 授权导入:


self.authorized_imports = list(set(BASE_BUILTIN_MODULES) | set(self.additional_authorized_imports))list(set(BASE_BUILTIN_MODULES) | set(self.additional_authorized_imports))


  • 将基础内置模块与额外的授权导入相结合,以定义允许导入的范围。


— 提示模板:


prompt_templates = prompt_templates or yaml.safe_load(or yaml.safe_load(
    importlib.resources.files("smolagents.prompts").joinpath("code_agent.yaml").read_text()
)


  • 从 YAML 文件中加载预定义的提示模板,确保智能体与大型语言模型(LLM)交互的一致性。


— Python 执行器:

  • 如果 use_e2b_executor 为 True,智能体使用 E2BExecutor 进行远程执行。
  • 否则,它默认使用 LocalPythonInterpreter。


— 验证:

  • 确保远程执行与管理智能体之间的兼容性。


if use_e2b_executor and len(self.managed_agents) > 0:
    raise Exception(
        f"You passed both {use_e2b_executor=} and some managed agents. Managed agents is not yet supported with remote code execution."
    )


系统提示初始化(initialize_system_prompt)

此方法生成系统提示,为大型语言模型(LLM)提供上下文和指令:


system_prompt = populate_template(
    self.prompt_templates["system_prompt"],"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)
        ),
    },
)


  • populate_template 函数使用提供的变量渲染 Jinja2 模板,确保系统提示的动态生成。
  • 系统提示包括有关可用工具、管理智能体和授权导入的详细信息。


步骤执行(step)

step 方法执行 ReAct 框架的一次迭代,包括代码生成、解析、执行和观察日志记录。以下是详细分解:


a. 生成模型输出

智能体通过将其内存转换为适合大型语言模型(LLM)的格式来准备输入消息。然后,它生成一个响应,其中包括下一个代码片段:


chat_message: ChatMessage = self.model(
    self.input_messages,
    stop_sequences=["<end_code>", "Observation:"],"<end_code>", "Observation:"],
    **additional_args,
)
model_output = chat_message.content


  • stop_sequences 参数确保大型语言模型(LLM)在遇到特定标记时停止生成输出。
  • 生成的代码片段存储在 model_output 中。


b. 解析代码

智能体使用 parse_code_blobs 函数解析生成的代码片段:


code_action = fix_final_answer_code(parse_code_blobs(model_output))


  • fix_final_answer_code 函数确保代码片段格式正确,以便执行。


c. 记录代码

智能体记录解析后的代码,以提高透明度:


self.logger.log_code(title="Executing parsed code:", content=code_action, level=LogLevel.INFO)"Executing parsed code:", content=code_action, level=LogLevel.INFO)


  • 该日志提供了正在执行的代码的可见性,使得调试和分析智能体的行为更加容易。


d. 执行代码

智能体使用配置的 Python 执行器执行解析后的代码:


output, execution_logs, is_final_answer = self.python_executor(
    code_action,
    self.state,
)


  • python_executor 运行代码片段,并返回输出、执行日志以及一个指示结果是否为最终答案的标志。


e. 处理错误

如果在执行过程中发生错误,智能体会记录错误并抛出异常:


if "Import of " in error_msg and " is not allowed" in error_msg:
    self.logger.log(
        "[bold red]Warning to user: Code execution failed due to an unauthorized import - Consider passing said import under `additional_authorized_imports` when initializing your CodeAgent.",
        level=LogLevel.INFO,
    )
raise AgentExecutionError(error_msg, self.logger)


  • 错误信息提供了解决未授权导入问题的指导。


f. 记录观察结果

智能体记录执行日志和输出,以提高透明度:


execution_outputs_console += [
    Text(
        f"{('Out - Final answer' if is_final_answer else 'Out')}: {truncated_output}",f"{('Out - Final answer' if is_final_answer else 'Out')}: {truncated_output}",
        style=(f"bold {YELLOW_HEX}" if is_final_answer else ""),
    ),
]
self.logger.log(Group(*execution_outputs_console), level=LogLevel.INFO)


  • 该日志在适用时突出显示最终答案,使得更容易识别解决方案。


g. 返回结果

如果结果是最终答案,智能体将其返回;否则,继续执行下一步:


return output if is_final_answer else None


用例

— 数学计算:

  • CodeAgent 非常适合需要复杂计算的任务,如解方程或进行统计分析。


— 数据分析:

  • 该智能体可以使用 Pandas 和 Matplotlib 等 Python 库处理和分析数据集、生成可视化图表并提取见解。


— 脚本自动化:

  • 该智能体可以通过生成和执行 Python 脚本来自动化重复性任务,如文件处理或 API 交互。


与专注于调用预定义工具的 ToolCallingAgent 不同,CodeAgent 强调执行动态生成的代码片段。CodeAgent 是构建能够通过 Python 代码生成和执行解决计算问题的智能体的强大工具。通过利用底层大型语言模型(LLM)和 Python 解释器的功能,它确保了操作的清晰性、精确性和可靠性。其与 ReAct 框架的集成以及全面的日志记录进一步增强了其功能,使开发人员能够为广泛的应用构建高效有效的解决方案。如文章开头图片所示,CodeAgent 通过减少与环境的交互和工具调用,可以实现更好的性能!


结论

HuggingFace 的 smolagents 库提供了一个健壮且灵活的框架,用于构建能够通过结构化推理、工具使用和动态规划解决复杂任务的智能体。三种主要的智能体类型——MultiStepAgent、CodeAgent 和 ToolCallingAgent——各有其独特用途,满足不同用例的需求,同时共享 ReAct 框架的共同基础。


MultiStepAgent 是该库的支柱,为需要行动和观察迭代周期的任务提供通用解决方案。其定期规划和适应的能力确保它始终专注于实现目标,使其适用于从代码调试到数据分析的广泛应用。


CodeAgent 专注于计算任务,利用 Python 代码生成和执行来动态解决问题。通过授权导入和远程执行等功能,它在安全性和灵活性之间取得了平衡,使开发人员能够精确地处理数学计算、数据分析和自动化工作流。


另一方面,ToolCallingAgent 在需要与外部工具或 API 进行结构化交互的场景中表现出色。通过生成类似 JSON 的工具调用,它确保了调用预定义操作的清晰性和可靠性,使其成为 API 集成、任务自动化和多工具工作流的理想选择。

文章来源:https://medium.com/@kargarisaac/exploring-the-smolagents-library-a-deep-dive-into-multistepagent-codeagent-and-toolcallingagent-03482a6ea18c
欢迎关注ATYUN官方公众号
商务合作及内容投稿请联系邮箱:bd@atyun.com
评论 登录
热门职位
Maluuba
20000~40000/月
Cisco
25000~30000/月 深圳市
PilotAILabs
30000~60000/年 深圳市
写评论取消
回复取消