LlamaIndex 是一种框架，用于构建使用 LLM 连接到您的数据的知识代理。此示例展示了如何为研究智能体构建多智能体工作流。在 LlamaIndex 中，Workflows 是智能体或多智能体系统的基本单元。
您需要 Gemini API 密钥。如果您还没有 API 密钥，可以在 Google AI Studio 中获取一个。
首先，安装所有必需的 LlamaIndex 库。LlamaIndex 在底层使用 google-genai 软件包。
pip install llama-index llama-index-utils-workflow llama-index-llms-google-genai llama-index-tools-google
在 LlamaIndex 中设置 Gemini 2.5 Pro
任何 LlamaIndex 代理的引擎都是一个 LLM，用于处理推理和文本处理。此示例使用 Gemini 2.5 Pro。请务必将 API 密钥设置为环境变量。
from llama_index.llms.google_genai import GoogleGenAI
llm = GoogleGenAI(model="gemini-2.5-pro")
构建工具
代理使用工具与外部世界互动，例如搜索网络或存储信息。LlamaIndex 中的工具可以是常规 Python 函数，也可以从预先存在的 ToolSpecs 中导入。Gemini 随附一个用于使用 Google 搜索的内置工具，我们在此处使用了该工具。
from google.genai import types
google_search_tool = types.Tool(
google_search=types.GoogleSearch()
)
llm_with_search = GoogleGenAI(
model="gemini-2.5-pro",
generation_config=types.GenerateContentConfig(tools=[google_search_tool])
)
现在，使用需要搜索的查询来测试 LLM 实例：
response = llm_with_search.complete("What's the weather like today in Biarritz?")
print(response)
研究代理将使用 Python 函数作为工具。您可以通过多种方式构建系统来执行此任务。在此示例中，您将使用以下内容：
search_web使用 Gemini 和 Google 搜索在网络上搜索有关指定主题的信息。record_notes将在网上找到的研究内容保存到状态，以便其他工具可以使用。write_report使用ResearchAgent找到的信息撰写报告review_report审核报告并提供反馈。
Context 类在代理/工具之间传递状态，每个代理都可以访问系统的当前状态。
from llama_index.core.workflow import Context
async def search_web(ctx: Context, query: str) -> str:
"""Useful for searching the web about a specific query or topic"""
response = await llm_with_search.acomplete(f"""Please research given this query or topic,
and return the result\n<query_or_topic>{query}</query_or_topic>""")
return response
async def record_notes(ctx: Context, notes: str, notes_title: str) -> str:
"""Useful for recording notes on a given topic."""
current_state = await ctx.store.get("state")
if "research_notes" not in current_state:
current_state["research_notes"] = {}
current_state["research_notes"][notes_title] = notes
await ctx.store.set("state", current_state)
return "Notes recorded."
async def write_report(ctx: Context, report_content: str) -> str:
"""Useful for writing a report on a given topic."""
current_state = await ctx.store.get("state")
current_state["report_content"] = report_content
await ctx.store.set("state", current_state)
return "Report written."
async def review_report(ctx: Context, review: str) -> str:
"""Useful for reviewing a report and providing feedback."""
current_state = await ctx.store.get("state")
current_state["review"] = review
await ctx.store.set("state", current_state)
return "Report reviewed."
构建多智能体助理
如需构建多代理系统，您需要定义代理及其互动。您的系统将有三个代理：
ResearchAgent会在网络上搜索有关指定主题的信息。WriteAgent使用ResearchAgent找到的信息撰写报告。ReviewAgent审核报告并提供反馈。
此示例使用 AgentWorkflow 类创建了一个多代理系统，该系统将按顺序执行这些代理。每个代理都会获取一个 system_prompt，其中包含有关其应执行的操作以及如何与其他代理协同工作的建议。
您可以选择使用 can_handoff_to 指定多代理系统可以与其他哪些代理对话，从而帮助该系统（如果不指定，该系统会尝试自行确定）。
from llama_index.core.agent.workflow import (
AgentInput,
AgentOutput,
ToolCall,
ToolCallResult,
AgentStream,
)
from llama_index.core.agent.workflow import FunctionAgent, ReActAgent
research_agent = FunctionAgent(
name="ResearchAgent",
description="Useful for searching the web for information on a given topic and recording notes on the topic.",
system_prompt=(
"You are the ResearchAgent that can search the web for information on a given topic and record notes on the topic. "
"Once notes are recorded and you are satisfied, you should hand off control to the WriteAgent to write a report on the topic."
),
llm=llm,
tools=[search_web, record_notes],
can_handoff_to=["WriteAgent"],
)
write_agent = FunctionAgent(
name="WriteAgent",
description="Useful for writing a report on a given topic.",
system_prompt=(
"You are the WriteAgent that can write a report on a given topic. "
"Your report should be in a markdown format. The content should be grounded in the research notes. "
"Once the report is written, you should get feedback at least once from the ReviewAgent."
),
llm=llm,
tools=[write_report],
can_handoff_to=["ReviewAgent", "ResearchAgent"],
)
review_agent = FunctionAgent(
name="ReviewAgent",
description="Useful for reviewing a report and providing feedback.",
system_prompt=(
"You are the ReviewAgent that can review a report and provide feedback. "
"Your feedback should either approve the current report or request changes for the WriteAgent to implement."
),
llm=llm,
tools=[review_report],
can_handoff_to=["ResearchAgent","WriteAgent"],
)
代理已定义，现在您可以创建 AgentWorkflow 并运行它。
from llama_index.core.agent.workflow import AgentWorkflow
agent_workflow = AgentWorkflow(
agents=[research_agent, write_agent, review_agent],
root_agent=research_agent.name,
initial_state={
"research_notes": {},
"report_content": "Not written yet.",
"review": "Review required.",
},
)
在工作流执行期间，您可以将事件、工具调用和更新流式传输到控制台。
from llama_index.core.agent.workflow import (
AgentInput,
AgentOutput,
ToolCall,
ToolCallResult,
AgentStream,
)
research_topic = """Write me a report on the history of the web.
Briefly describe the history of the world wide web, including
the development of the internet and the development of the web,
including 21st century developments"""
handler = agent_workflow.run(
user_msg=research_topic
)
current_agent = None
current_tool_calls = ""
async for event in handler.stream_events():
if (
hasattr(event, "current_agent_name")
and event.current_agent_name != current_agent
):
current_agent = event.current_agent_name
print(f"\n{'='*50}")
print(f"🤖 Agent: {current_agent}")
print(f"{'='*50}\n")
elif isinstance(event, AgentOutput):
if event.response.content:
print("📤 Output:", event.response.content)
if event.tool_calls:
print(
"🛠️ Planning to use tools:",
[call.tool_name for call in event.tool_calls],
)
elif isinstance(event, ToolCallResult):
print(f"🔧 Tool Result ({event.tool_name}):")
print(f" Arguments: {event.tool_kwargs}")
print(f" Output: {event.tool_output}")
elif isinstance(event, ToolCall):
print(f"🔨 Calling Tool: {event.tool_name}")
print(f" With arguments: {event.tool_kwargs}")
工作流程完成后，您可以打印报告的最终输出，以及审核代理的最终审核状态。
state = await handler.ctx.store.get("state")
print("Report Content:\n", state["report_content"])
print("\n------------\nFinal Review:\n", state["review"])
使用自定义工作流，实现更多功能
AgentWorkflow 是开始使用多智能体系统的好方法。但如果您需要更多控制权，该怎么办？您可以从头开始构建工作流。以下是您可能需要构建自己的工作流的一些原因：
- 更好地控制流程：您可以决定代理的确切路径。这包括创建循环、在特定时间点做出决策，或让代理并行处理不同的任务。
- 使用复杂数据：不要只使用简单的文本。借助自定义工作流，您可以使用更结构化的数据（例如 JSON 对象或自定义类）作为输入和输出。
- 处理不同的媒体类型：构建不仅能理解和处理文本，还能理解和处理图片、音频和视频的代理。
- 更智能的规划：您可以设计一个工作流，让代理在开始工作之前先创建详细的计划。这对于需要多个步骤的复杂任务非常有用。
- 启用自我修正功能：创建可以检查自己工作的代理。如果输出不够好，代理可以再次尝试，形成一个改进循环，直到结果完美为止。
如需详细了解 LlamaIndex 工作流，请参阅 LlamaIndex 工作流文档。