Meta开源的Llama官方Agent：Llama-Agentic-System深度解析

TLDR

• Meta 推出的 Llama as a System 将 Llama 3.1 模型转变为一个能够自主完成任务的智能代理，通过多步推理、工具使用和系统级安全机制，为构建更智能、更安全的 AI 应用开辟了新的可能性。

• Llama-Agentic-System 具备强大的功能，包括多步推理、工具使用（内置和零样本学习）以及系统级安全保障，能够处理更复杂的任务，并增强安全性。

• 本文将深入探讨 Llama as a System 的概念、架构、功能和应用场景，并提供详细的代码示例和使用指南。

引言

大型语言模型（LLM）的迅猛发展为人工智能领域带来了革命性的变化。从最初的文本生成工具，LLM 逐渐展现出理解复杂指令、执行多步任务的潜力。然而，如何将 LLM 安全、高效地应用于实际任务仍然是一个挑战。Meta 推出的 Llama 3.1 为解决这个问题提供了新的思路：将 Llama 作为一个系统，使其能够完成更复杂的任务，并增强其安全性。本文将详细介绍 Llama as a System 的概念、功能以及如何使用它构建 AI 应用。

Llama Agentic System 的核心功能

Llama Agentic System 的核心在于将 Llama 模型从一个简单的文本生成工具转变为一个能够自主完成任务的智能代理。它具备以下关键能力：

• 多步推理: 将复杂任务分解成多个步骤，并按照逻辑顺序执行。例如，要预订航班，模型需要先搜索航班信息，然后选择合适的航班，最后填写乘客信息并完成支付。

• 工具使用: 能够调用各种工具来完成任务，包括：

• 内置工具：如搜索引擎、代码解释器等。这些工具预先集成到系统中，模型可以直接调用。

• 零样本学习工具：可以通过上下文信息学习使用新的工具。这意味着即使模型之前没有见过某个工具，只要提供工具的使用说明，它就可以学会使用。

• 系统级安全: 将安全评估从模型级别提升到整个系统级别，通过 Llama Guard 等安全机制，确保系统在各种场景下的安全性。这包括对用户输入、模型输出以及工具调用进行安全检查和过滤，防止恶意攻击和不当内容的产生。

Llama Agentic System 架构解析

为了更好地理解 Llama Agentic System 的工作原理，让我们深入了解其架构。下图展示了 Llama Agentic System 的核心组件及其交互方式：

组件说明：

• User（用户）: 与 Llama Agentic System 进行交互的用户，发起任务指令并接收最终结果。

• Executor（执行器）: Llama Agentic System 的核心控制单元，负责接收用户输入，调用安全机制，并将任务分发给 LLM 或工具，最终将结果返回给用户。

• LLM（大型语言模型）: Llama 模型，负责理解任务、生成文本、选择合适的工具来执行任务，是系统的智能核心。

• Tools（工具）: 外部工具，如搜索引擎、代码解释器等，用于扩展 LLM 的功能，执行 LLM 无法直接完成的任务。

• Shields（安全机制）: 负责对用户输入、模型输出以及工具调用进行安全检查和过滤，确保系统安全可靠地运行。

工作流程：

1. 用户将任务指令和相关信息发送给 Executor。

2. Executor 调用安全机制对用户输入进行检查，确保输入内容安全无害。

3. Executor 将任务指令发送给 LLM，LLM 分析指令并选择合适的工具（如果需要）。

4. LLM 根据任务指令选择合适的工具，并将调用请求发送给 Executor。

5. Executor 调用安全机制对工具调用进行检查，防止潜在的安全风险。

6. Executor 调用相应的工具，并将工具返回的结果发送给 LLM，LLM 继续处理任务。

7. LLM 整合工具返回的结果，生成最终的回复，并发送给 Executor。

8. Executor 调用安全机制对模型输出进行检查，确保输出内容安全可靠。

9. Executor 将最终的回复发送给用户，完成任务。

Llama Agentic System 代码实现细节

Llama Agentic System 的核心代码位于 llama_agentic_system/agentic_system.py 文件中。以下是一些关键代码片段和解释：

1. AgentInstance 类

class AgentInstance(ShieldRunnerMixin):
def__init__(
        self,
        system_id:int,
        instance_config:AgenticSystemInstanceConfig,
        model:InstructModel,
        inference_api:Inference,
        builtin_tools:List[SingleMessageBuiltinTool],
        custom_tool_definitions:List[ToolDefinition],
        input_shields:List[ShieldBase],
        output_shields:List[ShieldBase],
        max_infer_iters:int=10,
        prefix_messages:Optional[List[Message]]=None,
    ):
# ... 初始化代码 ...

# 初始化内置工具字典
        self.tools_dict ={t.get_name(): t for t in builtin_tools}
# 初始化会话字典
        self.sessions ={}
# 初始化安全机制
ShieldRunnerMixin.__init__(
            self,
            input_shields=input_shields,
            output_shields=output_shields,
)

defcreate_session(self, name: str)->Session:
        # ... 创建会话代码 ...

AgentInstance 类表示一个 Llama Agentic System 实例，负责管理会话、执行推理、调用工具和安全机制等。它是 Llama Agentic System 的核心控制单元，协调各个组件协同工作。

• 它维护一个 tools_dict 字典，存储所有可用的内置工具，键为工具名，值为工具实例。

• 它维护一个 sessions 字典，存储所有会话，键为会话 ID，值为会话实例。

• 它使用 ShieldRunnerMixin 来管理输入和输出安全机制。

2. create_and_execute_turn 方法

    async defcreate_and_execute_turn(
        self, request: AgenticSystemTurnCreateRequest
    )->AsyncGenerator:
# ... 获取会话，构建历史消息 ...

# ... 创建轮次 ID，初始化参数 ...

# 发送轮次开始事件
yieldAgenticSystemTurnResponseStreamChunk(
            event=AgenticSystemTurnResponseEvent(
                payload=AgenticSystemTurnResponseTurnStartPayload(
                    turn_id=turn_id,
)
)
)

        steps =[]
        output_message =None
asyncfor chunk in self.run(
            turn_id=turn_id,
            input_messages=messages,
            temperature=params.temperature,
            top_p=params.top_p,
            stream=request.stream,
            max_gen_len=params.max_tokens,
):
# 处理推理过程中的事件流
# ...

# ... 断言 output_message 不为空 ...

# 创建轮次实例
        turn =Turn(
            turn_id=turn_id,
            session_id=request.session_id,
            input_messages=request.messages,
            output_message=output_message,
            started_at=start_time,
            completed_at=datetime.now(),
            steps=steps,
)
# 将轮次添加到会话中
        session.turns.append(turn)
# 发送轮次结束事件
yieldAgenticSystemTurnResponseStreamChunk(
            event=AgenticSystemTurnResponseEvent(
                payload=AgenticSystemTurnResponseTurnCompletePayload(
                    turn=turn,
)
)
        )

该方法负责创建一个新的会话轮次，并执行用户请求。它会根据用户输入调用 LLM 进行推理，并根据 LLM 的回复调用相应的工具。这个方法实现了 Llama Agentic System 的交互逻辑，处理用户输入并生成相应的输出。

• 它首先获取对应的会话，并构建包含历史消息的消息列表。

• 然后，它调用 self.run() 方法执行推理，并处理推理过程中产生的事件流。

• 最后，它创建 Turn 实例表示当前轮次，并发送轮次结束事件。

3. run 方法

    async defrun(
        self,
        turn_id:str,
        input_messages:List[Message],
        temperature:float,
        top_p:float,
        stream:bool=False,
        max_gen_len:Optional[int]=None,
    )->AsyncGenerator:
# 对用户输入调用安全机制
asyncfor res in self.run_shields_wrapper(
            turn_id, input_messages, self.input_shields,"user-input"
):
ifisinstance(res,bool):
return
else:
yield res

# 调用 _run 方法执行推理
asyncfor res in self._run(
            turn_id, input_messages, temperature, top_p, stream, max_gen_len
):
ifisinstance(res,bool):
return
elifisinstance(res,CompletionMessage):
                final_response = res
break
else:
yield res

# ... 断言 final_response 不为空 ...

# 对模型输出调用安全机制
        messages = input_messages +[final_response]
asyncfor res in self.run_shields_wrapper(
            turn_id, messages, self.output_shields,"assistant-output"
):
ifisinstance(res,bool):
return
else:
yield res

yield final_response

run 方法是 create_and_execute_turn 方法的核心，它实现了 Llama Agentic System 的推理逻辑，包括调用 LLM、执行工具、处理安全机制等。该方法是 Llama Agentic System 智能的核心，实现了多步推理、工具使用和安全保障等关键功能。

• 它首先对用户输入调用 run_shields_wrapper 方法进行安全检查。

• 然后，它调用 _run 方法执行实际的推理过程，并处理推理过程中产生的事件流。

• 最后，它对模型输出再次调用 run_shields_wrapper 方法进行安全检查。

4. _run 方法

    async def_run(
        self,
        turn_id:str,
        input_messages:List[Message],
        temperature:float,
        top_p:float,
        stream:bool=False,
        max_gen_len:Optional[int]=None,
    )->AsyncGenerator:
# 预处理消息，添加系统提示信息
        input_messages = preprocess_dialog(input_messages, self.prefix_messages)

        attachments =[]
        n_iter =0
whileTrue:
# ... 获取最后一条消息，打印消息内容 ...

            step_id =str(uuid.uuid4())
# 发送推理步骤开始事件
yieldAgenticSystemTurnResponseStreamChunk(
                event=AgenticSystemTurnResponseEvent(
                    payload=AgenticSystemTurnResponseStepStartPayload(
                        step_type=StepType.inference.value,
                        step_id=step_id,
)
)
)

# 构建推理请求
            req =ChatCompletionRequest(
                model=self.model,
                messages=input_messages,
                available_tools=self.instance_config.available_tools,
                stream=True,
                sampling_params=SamplingParams(
                    temperature=temperature,
                    top_p=top_p,
                    max_tokens=max_gen_len,
),
)

            tool_calls =[]
            content =""
            stop_reason =None
# 调用推理接口，处理推理结果
asyncfor chunk in self.inference_api.chat_completion(req):
# ... 处理推理结果 ...

# ... 处理推理结束原因 ...

# 创建 CompletionMessage 实例
            message =CompletionMessage(
                content=content,
                stop_reason=stop_reason,
                tool_calls=tool_calls,
)

# 发送推理步骤结束事件
yieldAgenticSystemTurnResponseStreamChunk(
                event=AgenticSystemTurnResponseEvent(
                    payload=AgenticSystemTurnResponseStepCompletePayload(
                        step_type=StepType.inference.value,
                        step_id=step_id,
                        step_details=InferenceStep(
                            step_id=step_id, turn_id=turn_id, model_response=message
),
)
)
)

# ... 处理推理结束条件 ...

# ... 处理模型调用工具 ...

            n_iter += 1

_run 方法是 run 方法的核心，它实现了 Llama Agentic System 的推理逻辑，包括调用 LLM、执行工具、处理安全机制等。

• 它首先预处理消息，添加系统提示信息。

• 然后，它进入一个循环，每次循环执行一次推理步骤，直到满足结束条件。

• 在每个推理步骤中，它首先构建推理请求，然后调用 self.inference_api.chat_completion() 方法执行推理，并处理推理结果。

• 如果模型需要调用工具，它会调用相应的工具，并将工具返回的结果添加到消息列表中。

Llama Agentic System Demo 示例

Llama Agentic System 提供了一些 Demo 示例，展示了如何使用它来完成实际任务。以下是一些示例：

1. 通货膨胀分析

• 代码路径: examples/scripts/inflation.py

• 功能描述: 该示例展示了如何使用 Llama Agentic System 来分析通货膨胀数据。它首先加载一个 CSV 文件，然后使用 LLM 来回答有关通货膨胀的问题，例如“哪一年以最高的通货膨胀结束？”、“是什么宏观经济形势导致了该时期如此高的通货膨胀？”等等。这个示例展示了 Llama Agentic System 如何处理结构化数据，并利用工具进行数据分析。

import asyncio
import fire
from llama_models.llama3_1.api.datatypes import*# noqa: F403
from custom_tools.ticker_data importTickerDataTool
from multi_turn import prompt_to_message, run_main

# 定义主函数
defmain(host: str, port: int, disable_safety: bool = False):
# 使用 asyncio 运行异步主函数
    asyncio.run(
        run_main(
[
UserMessage(
                    content=[
"Here is a csv, can you describe it ?",
Attachment(
                            url=URL(uri="file://examples/resources/inflation.csv"),
                            mime_type="text/csv",
),
],
),
                prompt_to_message("Which year ended with the highest inflation ?"),
                prompt_to_message(
"What macro economic situations that led to such high inflation in that period?"
),
                prompt_to_message("Plot average yearly inflation as a time series"),
                prompt_to_message(
"Using provided functions, get ticker data for META for the past 10 years ? plot percentage year over year growth"
),
                prompt_to_message(
"Can you take Meta's year over year growth data and put it in the same inflation timeseries as above ?"
),
],
            host=host,
            port=port,
            disable_safety=disable_safety,
            custom_tools=[TickerDataTool()],
)
)


if __name__ =="__main__":
    fire.Fire(main)

代码解读:

• 导入必要的库，包括 asyncio 用于异步操作，fire 用于命令行参数解析，llama_models 用于 Llama 模型相关功能，custom_tools 用于自定义工具，以及 multi_turn 用于多轮对话。

• 定义 main 函数，它是程序的入口点。

• 它使用 asyncio.run 运行异步函数 run_main。

• run_main 函数接收一系列用户消息和参数，包括主机地址、端口号、是否禁用安全机制以及自定义工具列表。

• 用户消息使用 UserMessage 类表示，可以包含文本、附件等内容。

• prompt_to_message 函数用于将文本提示转换为用户消息。

• TickerDataTool 是一个自定义工具，用于获取股票数据。

• 使用 fire.Fire(main) 将 main 函数绑定到命令行接口，方便用户使用命令行参数运行程序。

2. 旅行计划

• 代码路径: examples/scripts/vacation.py

• 功能描述: 该示例展示了如何使用 Llama Agentic System 来计划旅行。用户可以向系统提供一些旅行信息，例如目的地、时间等，系统会根据这些信息生成详细的旅行计划，包括景点推荐、路线规划、住宿建议等。这个示例展示了 Llama Agentic System 如何与用户进行多轮对话，并调用外部工具获取信息，最终完成复杂的任务。

import asyncio
import fire
from multi_turn import prompt_to_message, run_main

# 定义主函数
defmain(host: str, port: int, disable_safety: bool = False):
    asyncio.run(
        run_main(
[
                prompt_to_message(
"I am planning a trip to Switzerland, what are the top 3 places to visit?"
),
                prompt_to_message("What is so special about #1?"),
                prompt_to_message("What other countries should I consider to club?"),
                prompt_to_message("How many days should I plan for in each country?"),
],
            host=host,
            port=port,
            disable_safety=disable_safety,
)
)

# 当脚本作为主程序运行时，使用 fire 库执行 main 函数
if __name__ =="__main__":
    fire.Fire(main)