LlamaIndex多智能体系统开发
在这篇教程中，我们将使用LlamaIndex构建一个多智能体系统，从任何给定的文本生成高质量的Anki闪卡。
多智能体系统就像是一支由专业工人组成的团队，每个人都拥有自己的专长，共同完成复杂的任务。在这篇教程中，我们将使用LlamaIndex构建一个多智能体系统，从任何给定的文本生成高质量的Anki闪卡。
我们将从简单的开始，逐步增加更多的功能，这类似于你如何组建一个团队——从一个人开始，并根据需要逐渐添加更多的专家。
1、版本1：基本单智能体系统
在我们的第一个版本中，我们将创建一个只有一个智能体的简单系统，该智能体可以从文本中生成Anki闪卡。想象一下，这是你雇佣的第一个员工，他懂得如何创建教育内容。
1.1 架构
graph LR
Input[输入文本] -->|处理| QA[问答生成器]
QA -->|生成| Cards[闪卡]
style Input fill:#e1f5fe,stroke:#01579b
style QA fill:#e8f5e9,stroke:#2e7d32
style Cards fill:#fff3e0,stroke:#ef6c00
1.2 核心组件
让我们将代码分解为可管理的部分：
数据模型：首先，我们使用Pydantic定义我们的闪卡应该是什么样子：
from pydantic import BaseModel, Field
from typing import List
class QACard(BaseModel):
question: str
answer: str
extra: str
class Flashcard_model(BaseModel):
cards: List[QACard]
LLM设置：我们配置LlamaIndex以使用我们首选的语言模型：
def get_shared_llm():
"""返回一个供所有智能体共享的LLM实例。"""
return OpenAI(
model="gpt-4o-mini",
temperature=0,
api_base="http://127.0.0.1:4000/v1",
api_key="sk-test"
)
智能体创建：我们的问答生成器智能体专门用于创建闪卡：
def qa_generator_factory() -> OpenAIAgent:
system_prompt = """
你是一个专注于Anki闪卡生成的教育内容创作者。
你的任务是按照以下指南创建清晰简洁的闪卡：
1. 每张卡片应专注于一个特定的概念
2. 问题应清晰无歧义
3. 答案应简明但完整
4. 在额外字段中包括相关附加信息
5. 遵循最小信息原则
将每张卡片格式化为：
<card>
<question>您的问题在这里</question>
<answer>您的答案在这里</answer>
<extra>附加上下文、示例或解释</extra>
</card>
"""
return OpenAIAgent.from_tools(
[],
llm=get_shared_llm(),
system_prompt=system_prompt,
)
响应处理：我们使用转换函数将智能体的输出转换为结构化数据：
from tenacity import retry, stop_after_attempt, wait_exponential
from langchain.prompts import ChatPromptTemplate
from langchain.schema import ChatMessage
@retry(stop=stop_after_attempt(3), wait=wait_exponential(multiplier=1, min=4, max=10))
def transformer(message: str) -> dict:
chat_prompt_tmpl = ChatPromptTemplate(
message_templates=[
ChatMessage.from_str(message, role="user")
]
)
llm = get_shared_llm()
structured_data = llm.structured_predict(Flashcard_model, chat_prompt_tmpl)
return structured_data.model_dump()
1.3 主函数
以下是将所有内容组合在一起的方式：
def generate_anki_cards(input_text: str) -> dict:
# 创建智能体
agent = qa_generator_factory()
# 生成闪卡
response = agent.chat(
f"从以下文本生成Anki闪卡：\n\n{input_text}"
)
# 将响应转换为结构化数据
structured_response = transformer(str(response))
return structured_response
1.4 示例用法
让我们尝试使用关于RSI（相对强度指数）的示例文本：
sample_text = """
相对强度指数（RSI）是一种技术分析中的动量指标。
它衡量最近价格变动的速度和幅度，以评估超买或超卖状况。RSI显示为0到100之间的振荡器。
读数高于70通常表示超买状态，而读数低于30表示超卖状态。
"""
flashcards = generate_anki_cards(sample_text)
print(flashcards)
1.5 关键特性
- 单一职责：我们的智能体仅专注于创建闪卡
- 数据验证：Pydantic确保我们的数据结构正确
- 错误处理：重试机制有助于处理临时故障
- 结构化输出：一致的XML格式便于解析
- 类型安全：Python类型提示有助于早期捕获错误
1.6 当前限制
这个基本版本有一些限制，我们将在后续版本中解决这些问题：
- 没有质量控制（没有人审查卡片）
- 对复杂主题的理解有限
- 没有处理专门内容的方法（如代码示例）
- 没有记忆以前的消息
- 没有根据反馈改进卡片的方法
在下一个版本中，我们将添加第二个智能体来审查并改进生成的闪卡，使我们的系统更加强大。
2、版本2：双智能体系统
在这个版本中，我们将添加第二个智能体——审查者——以提高我们的闪卡质量。想象一下，就像有一个老师（问答生成器）创建闪卡，还有一个编辑者（审查者）来审查并改进它们。
2.1 架构
graph LR
Input[输入文本] -->|生成| QA[问答生成器]
QA -->|审查| Rev[审查者]
Rev -->|输出| Cards[最终闪卡]
style Input fill:#e1f5fe,stroke:#01579b
style QA fill:#e8f5e9,stroke:#2e7d32
style Rev fill:#f3e5f5,stroke:#7b1fa2
style Cards fill:#fff3e0,stroke:#ef6c00
2.2 新组件
让我们看看我们向系统中添加了什么：
智能体类型：首先，我们使用枚举定义我们的智能体类型：
from enum import Enum
class Speaker(str, Enum):
QA_GENERATOR = "问答生成器"
REVIEWER = "审查者"
审查者智能体：这个智能体专门负责提高闪卡质量：
def reviewer_factory() -> OpenAIAgent:
system_prompt = """
你是审查者智能体。你的任务是审查并完善Anki闪卡，确保它们遵循以下规则：
1. 每张卡片应测试一个信息片段
2. 问题必须：
- 简单直接
- 测试一个事实
- 在适当的情况下使用填空格式
3. 答案必须：
- 简短准确
- 仅限于必要信息
4. 额外字段必须包括：
- 详细的解释
- 示例
- 上下文
"""
return OpenAIAgent.from_tools(
[],
llm=get_shared_llm(),
system_prompt=system_prompt,
)
状态管理：我们跟踪闪卡创建的进度：
def get_initial_state(text: str) -> dict:
return {
"input_text": text,
"qa_cards": "",
"review_status": "待审查"
}
2.3 主函数
以下是将两个智能体结合起来的方式：
def generate_anki_cards(input_text: str) -> dict:
# 初始化状态
state = get_initial_state(input_text)
# 步骤1：生成初始卡片
generator = qa_generator_factory()
response = generator.chat(
f"从以下文本生成Anki闪卡：\n\n{state['input_text']}"
)
state["qa_cards"] = str(response)
# 步骤2：审查并改进卡片
reviewer = reviewer_factory()
review_response = reviewer.chat(
f"审查并改进这些闪卡：\n\n{state['qa_cards']}"
)
state["qa_cards"] = str(review_response)
# 将最终卡片转换为结构化数据
return transformer(state["qa_cards"])
2.4 示例用法
让我们看看它与我们的RSI示例一起工作的情况：
sample_text = """
相对强度指数（RSI）是一种技术分析中的动量指标。
它衡量最近价格变动的速度和幅度，以评估超买或超卖状况。RSI显示为0到100之间的振荡器。
读数高于70通常表示超买状态，而读数低于30表示超卖状态。
"""
flashcards = generate_anki_cards(sample_text)
``````markdown
<cards>
<card>
<question>RSI (相对强度指数)是技术分析中的一个_____指标。</question>
<answer>动量</answer>
<extra>
关于RSI的关键点：
- 衡量价格变化的速度和幅度
- 规模：0到100
- 超买：高于70
- 超卖：低于30
</extra>
</card>
</cards>
2.5 主要改进
- 质量控制：审查者代理确保更好的卡片质量。
- 更好的格式：Cloze 删除式更有效的学习。
- 更丰富的上下文：额外字段内容更加详细。
- 进度跟踪：状态管理显示卡片状态。
- 一致的结构：保持XML格式。
2.6 当前限制
虽然比版本1更好，但该系统仍然存在一些限制：
- 没有办法分析复杂主题。
- 基本的两步流程（可以更复杂）。
- 没有记忆以前的消息。
- 错误处理有限。
- 没有办法处理专业化内容（如代码）。
在版本3中，我们将添加更多代理来处理这些限制并创建一个更强大的系统。
3、版本3：多代理乐团
在这个版本中，我们通过启用多个代理有效协作来增强我们的系统。每个代理都有特定的角色，并且它们一起工作以创建高质量的闪存卡。协调者代理在管理这种协作方面发挥着关键作用，确保每个代理都能有效贡献。此外，主题分析器有助于将复杂主题分解为更简单的部分，提供内容的结构化概述。
3.1 多代理协作如何工作
在一个多代理系统中，每个代理就像一个具有特定专长的团队成员。以下是他们如何协作的方式：
- 专门角色：每个代理被设计为执行特定任务。例如，问答生成器创建闪存卡，而审查者确保其质量。
- 顺序处理：代理按顺序工作，每个代理都建立在前一个代理的输出基础上。这确保最终产品经过精炼且全面。
- 动态决策制定：协调者代理根据当前状态决定哪个代理应该运行。这允许系统根据内容的需求动态适应。
- 共享上下文：内存系统在整个交互过程中维护上下文，使代理能够基于先前的输出做出明智的决策。
3.2 引入协调者代理
协调者是我们多代理系统的管理者。它协调工作流程，确保每个代理都能有效贡献。以下是它的运作方式：
协调者的角色：
- 决策制定：协调者评估当前状态并决定哪个代理应运行下一步。它确保工作流程高效且所有必要任务都已完成。
- 灵活性：协调者可以在任何时候选择任何代理，允许灵活和适应性的过程。如果需要，它可以多次运行代理或跳过步骤，如果内容已经完成。
- 结束条件：协调者决定何时完成过程，确保最终闪存卡全面且高质量。
以下是协调者的实现方式：
def orchestrator_factory(state: dict) -> OpenAIAgent:
system_prompt = f"""
你是一个协调者代理。你的任务是协调所有代理之间的互动，以创建高质量的闪存卡。
当前状态：
{pformat(state, indent=2)}
可用代理：
* 主题分析器 - 分解复杂主题
* 问答生成器 - 创建闪存卡
* 审查者 - 改进卡片质量
决策指南：
- 首先使用主题分析器进行主题分解
- 使用问答生成器创建卡片
- 使用审查者细化卡片
- 选择“结束”当卡片准备好时
"""
return OpenAIAgent.from_tools(
[],
llm=get_shared_llm(),
system_prompt=system_prompt,
)
3.3 引入主题分析器代理
主题分析器负责将复杂主题分解为可管理的部分。它提供了内容的结构化概述，有助于创建更集中和全面的闪存卡。
主题分析器的角色：
- 内容分解：主题分析器从输入文本中识别关键主题和子主题，帮助构建内容结构。
- 层次结构构建：它创建主题的层次结构，使理解不同概念之间的关系更容易。
- 应用建议：代理建议与主题相关的实际应用或案例研究，丰富闪存卡的实用见解。
以下是主题分析器的实现方式：
def topic_analyzer_factory(state: dict) -> OpenAIAgent:
system_prompt = f"""
你是一个主题分析器代理。你的任务是分析给定文本并为闪存卡创建识别关键主题。
当前状态：
{pformat(state, indent=2)}
指令：
1. 识别主要概念和子概念
2. 创建主题的结构列表
3. 提议实际应用
输出格式：
<topics>
<topic>
<name>主要主题名称</name>
<subtopics>
<subtopic>子主题1</subtopic>
<subtopic>子主题2</subtopic>
</subtopics>
</topic>
</topics>
"""
return OpenAIAgent.from_tools(
[],
llm=get_shared_llm(),
system_prompt=system_prompt,
verbose=True
)
3.4 主函数
主函数现在包括协调者，它动态决定工作流程：
def generate_anki_cards(input_text: str) -> dict:
# 初始化状态和内存
state = get_initial_state(input_text)
memory = setup_memory()
while True:
# 获取当前聊天历史记录
current_history = memory.get()
# 让协调者决定下一步
orchestrator = orchestrator_factory(state)
next_agent = str(orchestrator.chat(
"根据当前状态决定运行哪个代理。",
chat_history=current_history
)).strip().strip('"').strip("'")
print(f"\n协调者选定：{next_agent}")
if next_agent == "结束":
print("\n协调者决定结束过程")
break
# 执行选定的代理
try:
if next_agent == Speaker.TOPIC_ANALYZER.value:
analyzer = topic_analyzer_factory(state)
response = analyzer.chat(
f"分析这段文本以获取闪存卡主题：\n\n{state['input_text']}",
chat_history=current_history
)
state["topics"] = str(response)
print("\n主题分析结果：")
print(state["topics"])
elif next_agent == Speaker.QA_GENERATOR.value:
generator = qa_generator_factory()
response = generator.chat(
f"为这个主题生成闪存卡：\n\n{state['topics']}",
chat_history=current_history
)
state["qa_cards"] = str(response)
state["review_status"] = "需要审查"
print("\n生成的卡片：")
print(state["qa_cards"])
elif next_agent == Speaker.REVIEWER.value:
reviewer = reviewer_factory()
response = reviewer.chat(
f"审查这些闪存卡：\n\n{state['qa_cards']}",
chat_history=current_history
)
state["qa_cards"] = str(response)
state["review_status"] = "已审查"
print("\n审查过的卡片：")
print(state["qa_cards"])
# 更新内存中的新交互
memory.put(ChatMessage(role="assistant", content=str(response)))
print(f"\n更新内存与{next_agent}的响应")
except Exception as e:
print(f"\n{next_agent}中的错误：{str(e)}")
continue
# 将最终卡片转换为结构化数据
final_cards = transformer(state["qa_cards"])
return final_cards
3.5 示例用法
以下是协调系统如何与我们的RSI示例一起工作的例子：
if __name__ == "__main__":
sample_text = """
相对强度指数（RSI）是技术分析中使用的动量指标。
它衡量近期价格变化的速度和幅度，以评估超买或超卖情况。RSI显示为0到100的振荡器。
读数高于70通常表示超买状况，而读数低于30表示超卖状况。RSI还可以帮助识别趋势线、背离和失败波动，这些可能在基本价格图表上不明显。
"""
flashcards = generate_anki_cards(sample_text)
print("\n生成、分析和审查的闪存卡：")
print(flashcards)
进入全屏模式 退出全屏模式 ted Workflow: 管理员交互，确保流程顺畅高效。 2. 主题分析: 提供更深入的内容结构理解，生成更全面的闪卡。 3. 持久记忆: 跨交互保持上下文，实现更明智的决策。 4. 灵活执行: 根据当前状态动态选择代理，适应内容需求。 5. 分层处理: 将主题分解为可管理的部分，使创建详细准确的闪卡更容易。
3.6 限制
尽管此版本更为复杂，但仍有一些改进空间：
- 没有专门处理代码示例的方法
- 格式化选项有限
- 基本错误处理
在版本4中，我们将引入代码和额外字段专家代理和格式化代理来处理代码示例并确保一致的格式。我们还将改进错误处理并添加对不同内容类型的支持。
4、版本4：增强功能
在此版本中，我们引入了两个新代理：代码和额外字段专家以及格式化代理。这些代理分别处理代码示例和确保一致的格式。我们还改进了错误处理并添加了对不同内容类型的支持。
4.1 架构
graph TB
Input[输入文本] -->|开始| Orch[协调器]
Orch -->|分析| TA[主题分析器]
Orch -->|处理代码| Code[代码专家]
Orch -->|生成| QA[问答生成器]
Orch -->|格式化| Format[格式化器]
Orch -->|审查| Rev[审查者]
TA -->|主题| 决策{继续？}
Code -->|结果| 决策
QA -->|结果| 决策
Format -->|结果| 决策
Rev -->|结果| 决策
决策 -->|是| Orch
决策 -->|否| Final[最终闪卡]
style Input fill:#e1f5fe,stroke:#01579b
style Orch fill:#fff3e0,stroke:#ef6c00
style TA fill:#e8f5e9,stroke:#2e7d32
style Code fill:#e8f5e9,stroke:#2e7d32
style QA fill:#bbdefb,stroke:#1976d2
style Format fill:#bbdefb,stroke:#1976d2
style Rev fill:#f3e5f5,stroke:#7b1fa2
style Final fill:#fff3e0,stroke:#ef6c00
style 决策 fill:#ffecb3,stroke:#ffa000
4.2 新组件
我们扩展了代理类型以包括新的角色：
class Speaker(str, Enum):
QA_GENERATOR = "问答生成器"
REVIEWER = "审查者"
TOPIC_ANALYZER = "主题分析器"
ORCHESTRATOR = "协调器"
CODE_AND_EXTRA_FIELD_EXPERT = "代码和额外字段专家"
FORMATTER = "格式化器"
该代理通过添加相关代码片段和详细的额外内容来增强闪卡：
def code_and_extra_field_expert_factory() -> OpenAIAgent:
system_prompt = """
你是一个代码和额外字段专家代理。你的任务是通过添加相关的代码片段和全面的额外内容来增强Anki闪卡。
指令：
1. 添加清晰简洁的代码示例，说明关键概念
2. 确保代码片段注释清晰易懂
3. 在额外字段中提供：
- 代码片段的逐步解释
- 常见用例和场景
- 潜在陷阱和边缘情况
- 最佳实践和优化建议
4. 使用适当的markdown格式化代码块
5. 包含相关文档链接
6. 确保解释对15岁的人来说清晰易懂
"""
return OpenAIAgent.from_tools(
[],
llm=get_shared_llm(),
system_prompt=system_prompt,
)
该代理确保闪卡格式正确：
def formatter_agent_factory() -> OpenAIAgent:
system_prompt = """
你是格式化代理。你的任务是确保闪卡中的有效XML结构和markdown格式。
格式规则：
1. 维护有效的XML结构
2. 正确转义特殊字符
3. 使用适当的语言标签格式化代码块
4. 使用一致的缩进
5. 确保markdown兼容性
6. 保留提供的代码片段完全不变
7. 正确处理嵌套结构
"""
return OpenAIAgent.from_tools(
[],
llm=get_shared_llm(),
system_prompt=system_prompt,
)
4.3 增强的错误处理和验证
我们通过重试和验证改进了错误处理：
@retry(stop=stop_after_attempt(3), wait=wait_exponential(multiplier=1, min=4, max=10))
def validate_and_transform(message: str) -> dict:
try:
# 转换为结构化数据
chat_prompt_tmpl = ChatPromptTemplate(
message_templates=[
ChatMessage.from_str(message, role="用户")
]
)
structured_data = get_shared_llm().structured_predict(
Flashcard_model,
chat_prompt_tmpl
)
return structured_data.model_dump()
except ET.ParseError as e:
print(f"XML验证错误：{str(e)}")
raise TryAgain
except Exception as e:
print(f"转换错误：{str(e)}")
raise TryAgain
4.4 示例使用
以下是包含代码的增强系统的工作方式示例：
if __name__ == "__main__":
sample_text = """
要在Python中计算RSI（相对强度指数），通常会使用pandas-ta或ta库等技术分析库。RSI是根据指定周期（通常是14个周期）的平均收益和损失计算得出的。以下是如何实现它的方法：
1. 使用ta库：
```
python
import pandas as pd
import ta
# 假设你在DataFrame中有价格数据
df['RSI'] = ta.momentum.RSIIndicator(
close=df['close'],
window=14
).rsi()
```
2. 手动实现：
```
python
def calculate_rsi(data, periods=14):
delta = data.diff()
gain = (delta.where(delta > 0, 0)).rolling(
window=periods).mean()
loss = (-delta.where(delta < 0, 0)).rolling(
window=periods).mean()
rs = gain / loss
return 100 - (100 / (1 + rs))
```
"""
flashcards = generate_anki_cards(sample_text)
print("带有代码示例的生成闪卡：")
print(flashcards)
4.5 关键改进
- 代码处理：专门处理代码示例和解释的代理
- 一致格式化：专门的格式化代理
- 强大的错误处理：改进的验证和重试机制
- 丰富的额外内容：全面的解释和最佳实践
- Markdown支持：代码块和文本的正确格式化
5、结束语
在整个教程中，我们构建了一个用于生成Anki闪卡的复杂多代理系统，从简单的单代理实现逐步发展到复杂的协调系统。让我们回顾每个版本的关键发展：
5.1 系统演变
版本1：基本单代理
- 建立核心数据结构
- 实现基本闪卡生成
- 设置错误处理基础
版本2：两代理交互
- 通过审查过程增加了质量控制
- 引入了状态管理
- 实现了迭代细化
版本3：多代理管弦乐
- 添加了协调层
- 实现了主题分析
- 通过内存系统增强了上下文管理
版本4：增强功能
- 添加了专门的代码处理
- 实现了一致的格式化
- 改进了错误处理和验证
5.2 关键收获
模块化设计
- 每个代理都有特定的责任
- 易于添加新代理或修改现有代理
- 清晰的责任分离
稳健架构
- 代理之间的状态管理
- 错误处理和重试机制
- 内存系统用于上下文保存
质量控制
- 多层次的审查和细化
- 一致的格式和结构
- 丰富的带代码示例和解释的内容
5.3 未来增强
虽然我们的系统已经相当强大，但在以下几个方面还有进一步改进的空间：
并行处理
- 实现并发卡片生成
- 添加批量处理功能
- 优化大规模操作
高级功能
- 图像处理和处理
- 多语言支持
- 音频内容集成
- 交互式示例
学习能力
- 反馈整合
- 质量指标跟踪
- 自适应提示优化
集成选项
- API端点
- Web界面
- 直接Anki集成
- 卡片版本控制
6、结束语
构建一个多代理系统需要仔细考虑代理交互、状态管理和错误处理。本教程展示了如何逐步构建这样的系统，从基本组件逐步增加复杂度。
最终系统展示了结合多个具有各自专长的专业代理的强大和灵活性。虽然重点在于Anki闪卡生成，但这里展示的模式和原则可以应用于许多其他多代理应用。
记住，成功多代理系统的要点在于：
- 清晰的代理职责
- 健壮的通信模式
- 全面的错误处理
- 灵活的状态管理一致的输出格式
通过遵循这些原则和本教程中展示的模式，您可以为各种应用程序构建自己的多智能体系统，而不仅仅是闪卡生成。
原文链接：Building a Multi-Agent Framework from Scratch with LlamaIndex
汇智网翻译整理，转载请标明出处
