引入 llama-agents：一个用于构建生产级多智能体 AI 系统的强大框架

我们很高兴宣布 llama-agents 的 Alpha 版本发布，这是一个新的开源框架，旨在简化构建、迭代和部署多智能体 AI 系统的过程，并将您的智能体转化为生产级微服务。无论您是处理复杂的问答系统、协作式 AI 助手还是分布式 AI 工作流，llama-agents 都提供了将您的想法变为现实所需的工具和结构。

llama-agents 的关键特性

1. 分布式服务导向架构：LlamaIndex 中的每个智能体都可以成为其独立的微服务，由可完全定制的 LLM 驱动的控制平面进行编排，该控制平面负责任务路由和分配。
2. 通过标准化 API 接口通信：使用中心控制平面编排器在智能体之间进行接口。使用消息队列在智能体之间传递消息。
3. 定义智能体驱动和显式的编排流程：开发者可以灵活地直接定义智能体之间的交互序列，或者交给一个“智能体编排器”来决定哪些智能体与任务相关。
4. 易于部署：独立启动、扩展和监控每个智能体及其控制平面。
5. 可扩展性和资源管理：使用我们内置的可观测性工具来监控系统和每个单独智能体服务的质量和性能。

让我们深入了解如何开始使用 llama-agents 构建您自己的多智能体系统。

llama-agents 入门

首先，使用 pip 安装该框架

pip install llama-agents llama-index-agent-openai

基本系统设置

这里是一个使用 llama-agents 设置基本多智能体系统的简单示例。首先，我们将引入依赖项并设置我们的控制平面，其中包含我们由 LLM 驱动的编排器

import dotenv
dotenv.load_dotenv() # our .env file defines OPENAI_API_KEY
from llama_agents import (
    AgentService,
    ControlPlaneServer,
    SimpleMessageQueue,
    AgentOrchestrator,
)
from llama_index.core.agent import FunctionCallingAgentWorker
from llama_index.core.tools import FunctionTool
from llama_index.llms.openai import OpenAI
import logging

# turn on logging so we can see the system working
logging.getLogger("llama_agents").setLevel(logging.INFO)

# Set up the message queue and control plane
message_queue = SimpleMessageQueue()
control_plane = ControlPlaneServer(
    message_queue=message_queue,
    orchestrator=AgentOrchestrator(llm=OpenAI()),
)

接下来，我们使用 LlamaIndex 现有的抽象创建工具，将这些工具提供给一个智能体，并将该智能体转化为独立的微服务

# create a tool
def get_the_secret_fact() -> str:
    """Returns the secret fact."""
    return "The secret fact is: A baby llama is called a 'Cria'."

tool = FunctionTool.from_defaults(fn=get_the_secret_fact)

# Define an agent
worker = FunctionCallingAgentWorker.from_tools([tool], llm=OpenAI())
agent = worker.as_agent()

# Create an agent service
agent_service = AgentService(
    agent=agent,
    message_queue=message_queue,
    description="General purpose assistant",
    service_name="assistant",
)

最后，我们启动服务和控制平面。请注意，这里我们使用一个辅助函数来运行单个查询并通过系统然后退出；接下来我们将展示如何将其部署到生产环境。

# Set up the launcher for local testing
from llama_agents import LocalLauncher

launcher = LocalLauncher(
    [agent_service],
    control_plane,
    message_queue,
)

# Run a single query through the system
result = launcher.launch_single("What's the secret fact?")
print(result)

部署您的多智能体系统

在本地测试系统后，您可以将其部署为一组服务，以供实际生产使用。以下是您可能设置的方式。这与之前的示例类似，但我们添加了第二个智能体服务并使用了不同的启动器。让我们再次引入依赖项并设置我们的控制平面

import dotenv
dotenv.load_dotenv()
from llama_agents import (
    AgentService,
    AgentOrchestrator,
    ControlPlaneServer,
    SimpleMessageQueue,
)

from llama_index.core.agent import FunctionCallingAgentWorker
from llama_index.core.tools import FunctionTool
from llama_index.llms.openai import OpenAI
import logging

# change logging level to enable or disable more verbose logging
logging.getLogger("llama_agents").setLevel(logging.INFO)

# create our multi-agent framework components
message_queue = SimpleMessageQueue()
control_plane = ControlPlaneServer(
    message_queue=message_queue,
    orchestrator=AgentOrchestrator(llm=OpenAI()),
)

然后像之前一样，我们创建一个工具和一个智能体，不过这次我们将添加第二个智能体

# create a tool
def get_the_secret_fact() -> str:
    """Returns the secret fact."""
    return "The secret fact is: A baby llama is called a 'Cria'."

tool = FunctionTool.from_defaults(fn=get_the_secret_fact)

# create our agents
worker1 = FunctionCallingAgentWorker.from_tools([tool], llm=OpenAI())
worker2 = FunctionCallingAgentWorker.from_tools([], llm=OpenAI())
agent1 = worker1.as_agent()
agent2 = worker2.as_agent()

我们将这些智能体转化为服务

agent_server_1 = AgentService(
    agent=agent1,
    message_queue=message_queue,
    description="Useful for getting the secret fact.",
    service_name="secret_fact_agent",
    host="localhost",
    port=8003
)
agent_server_2 = AgentService(
    agent=agent2,
    message_queue=message_queue,
    description="Useful for getting random dumb facts.",
    service_name="dumb_fact_agent",
    host="localhost",
    port=8004
)

最后，我们将每个服务作为独立的智能体启动。这里我们都在一个脚本中完成，但每个都可以是完全独立的服务，独立启动和扩展

from llama_agents import ServerLauncher, CallableMessageConsumer

# Additional human consumer
def handle_result(message) -> None:
    print(f"Got result:", message.data)


# the final result is published to a "human" consumer
# so we define one to handle it!
human_consumer = CallableMessageConsumer(
    handler=handle_result, message_type="human"
)

# Define Launcher
launcher = ServerLauncher(
    [agent_server_1, agent_server_2],
    control_plane,
    message_queue,
    additional_consumers=[human_consumer]
)

launcher.launch_servers()

实时监控

我们的多智能体系统最酷的调试功能之一是内置的智能体监视器。您可以这样启动它

llama-agents monitor --control-plane-url http://127.0.0.1:8000

启动后，您会得到一个直观、点选式的终端应用程序。您可以看到两个智能体都在运行，底部可以注入一个任务，例如查询“秘密事实是什么？”。您将获得一个任务 ID，点击该 ID 即可查看结果。

构建一个查询重写 RAG 系统

让我们看一个更复杂的示例：一个查询重写 RAG 系统。该系统将重写用户查询以改进检索，然后使用重写后的查询对文档执行 RAG。

此示例演示了如何创建一个更复杂的系统，该系统结合了查询重写和 RAG 以提高问答能力。有关更详细的解释，请参阅此 notebook。

import dotenv
dotenv.load_dotenv() # our .env defines OPENAI_API_KEY
from llama_index.core import VectorStoreIndex, Document
from llama_index.core.agent import FnAgentWorker
from llama_index.core import PromptTemplate
from llama_index.core.query_pipeline import QueryPipeline
from llama_index.core.query_engine import RetrieverQueryEngine
from llama_agents import (
    AgentService,
    ControlPlaneServer,
    SimpleMessageQueue,
    PipelineOrchestrator,
    ServiceComponent,
)
from llama_agents.launchers import LocalLauncher
from llama_index.llms.openai import OpenAI
import logging

# change logging level to enable or disable more verbose logging
logging.getLogger("llama_agents").setLevel(logging.INFO)

# Load and index your document
docs = [Document(text="The rabbit is a small mammal with long ears and a fluffy tail. His name is Peter.")]
index = VectorStoreIndex.from_documents(docs)

# Define a query rewrite agent
HYDE_PROMPT_STR = (
    "Please rewrite the following query to include more detail:\n{query_str}\n"
)
HYDE_PROMPT_TMPL = PromptTemplate(HYDE_PROMPT_STR)

def run_hyde_fn(state):
    prompt_tmpl, llm, input_str = (
        state["prompt_tmpl"],
        state["llm"],
        state["__task__"].input,
    )
    qp = QueryPipeline(chain=[prompt_tmpl, llm])
    output = qp.run(query_str=input_str)
    state["__output__"] = str(output)
    return state, True

hyde_agent = FnAgentWorker(
    fn=run_hyde_fn,
    initial_state={"prompt_tmpl": HYDE_PROMPT_TMPL, "llm": OpenAI()}
).as_agent()

# Define a RAG agent
def run_rag_fn(state):
    retriever, llm, input_str = (
        state["retriever"],
        state["llm"],
        state["__task__"].input,
    )
    query_engine = RetrieverQueryEngine.from_args(retriever, llm=llm)
    response = query_engine.query(input_str)
    state["__output__"] = str(response)
    return state, True

rag_agent = FnAgentWorker(
    fn=run_rag_fn,
    initial_state={"retriever": index.as_retriever(), "llm": OpenAI()}
).as_agent()

# Set up the multi-agent system
message_queue = SimpleMessageQueue()

query_rewrite_service = AgentService(
    agent=hyde_agent,
    message_queue=message_queue,
    description="Query rewriting service",
    service_name="query_rewrite",
)

rag_service = AgentService(
    agent=rag_agent,
    message_queue=message_queue,
    description="RAG service",
    service_name="rag",
)

# Create the pipeline
pipeline = QueryPipeline(chain=[
    ServiceComponent.from_service_definition(query_rewrite_service),
    ServiceComponent.from_service_definition(rag_service),
])
orchestrator = PipelineOrchestrator(pipeline)

control_plane = ControlPlaneServer(
    message_queue=message_queue,
    orchestrator=orchestrator,
)

# Set up the launcher
launcher = LocalLauncher(
    [query_rewrite_service, rag_service],
    control_plane,
    message_queue,
)

# Run a query
result = launcher.launch_single("Tell me about rabbits")
print(result)

公开路线图

这是一个 Alpha 版本，这意味着我们非常期待您对功能提出反馈，以便更好地帮助您在生产环境中构建多智能体系统！我们创建了一个公开路线图，展示了我们未来的规划。我们正在积极征集公众反馈，了解哪些地方有效，哪些地方无效。

立即深入探索！

llama-agents 提供了一个强大、灵活的框架，用于构建复杂的多智能体 AI 系统。无论您是原型化新想法还是扩展到生产环境，llama-agents 都提供了将您的 AI 愿景变为现实所需的工具。查阅仓库了解更多信息，特别是我们的示例库。

我们很高兴看到社区使用 llama-agents 构建出什么精彩的应用。编程愉快！