LLamaIndex 多Agent工作流配合使用

最新推荐文章于 2025-05-30 22:44:54 发布

mousegavin

最新推荐文章于 2025-05-30 22:44:54 发布

阅读量300

点赞数 3

文章标签： python langchain

本文链接：https://blog.csdn.net/mousegavin/article/details/147743343

版权

在此之前一直在使用创建单个代理。但也旨在支持多代理系统，其中多个代理协作完成您的任务，并根据需要将控制权移交给彼此。AgentWorkflow

我们的系统将有三个代理：

A 将在 Web 上搜索有关给定主题的信息。ResearchAgent
A 将使用 ResearchAgent 找到的信息编写报告。WriteAgent
A 将审阅报告并提供反馈。ReviewAgent

创建一个多代理系统，该系统将按顺序执行这些代理。有很多方法可以构建一个系统来执行这项任务。在此示例中，将使用一些工具来帮助研究和编写过程。

一个在 Web 上搜索有关给定主题的信息的工具（我们将使用 Tavily，就像我们在前面的示例中所做的那样）web_search
一个将在 Web 上找到的研究保存到 state 以便其他工具可以使用它的工具（参见 state management 来提醒自己这是如何运作的）record_notes
一个工具，用于使用write_reportResearchAgent
用于查看报告和提供反馈的工具。review_report

利用 Context 类，我们可以在 Agent 之间传递状态，并且每个 Agent 都可以访问系统的当前状态。

创建第一个代理，它将使用该工具在 Web 上搜索信息，并使用该工具将这些注释保存到状态以供其他代理使用。这里需要注意的关键语法元素是： * 的，用于向其他代理标识代理，我们稍后会看到 * 其他代理使用它来决定将控制权移交给下一个谁 * 的，用于定义代理的行为 * 是代理可以将其控制权交给的代理名称的可选列表。默认情况下，它将能够将控制权交给任何其他代理。

# 定义一个函数，模拟搜索函数
async def search_web(query: str) -> dict[str, str]:
    """根据查询返回相关信息。"""
    print(f"调用search: {query}")
    return {
        "query": query,
        "result": """
        Python 是一种解释型、面向对象、动态数据类型的高级程序设计语言。
        Python 由 Guido van Rossum 于 1989 年底发明，第一个公开发行版发行于 1991 年。
        像 Perl 语言一样, Python 源代码同样遵循 GPL(GNU General Public License) 协议。
        官方宣布，2020 年 1 月 1 日， 停止 Python 2 的更新。
        Python 2.7 被确定为最后一个 Python 2.x 版本。
        """
    }

research_agent = FunctionAgent(
        name="ResearchAgent",
        # 修改描述为中文
        description="用于在网络上搜索给定主题的信息并记录相关笔记。",
        # 修改系统提示词为中文
        system_prompt=(
            "你是研究代理，能够在网络上搜索给定主题的信息并记录相关笔记。 "
            "一旦笔记记录完成且你感到满意，你应该将控制权移交给撰写代理，让其撰写关于该主题的报告。"
        ),
        llm=llm,
        tools=[search_web,record_notes],
        can_handoff_to=["WriteAgent"]
)

其他两个代理的定义类似

async def record_notes(ctx: Context, notes: str, notes_title: str) -> str:
    """Useful for recording notes on a given topic."""
    print(f"调用record_notes: {notes}, {notes_title}")
    current_state = await ctx.get("state")
    if "research_notes" not in current_state:
        current_state["research_notes"] = {}
    current_state["research_notes"][notes_title] = notes
    await ctx.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."""
    print(f"调用write_report: {report_content}")
    current_state = await ctx.get("state")
    current_state["report_content"] = report_content
    await ctx.set("state", current_state)
    return "Report written."

async def review_report(ctx: Context, review: str) -> str:
    """Useful for reviewing a report and providing feedback."""
    print(f"调用review_report: {review}")
    current_state = await ctx.get("state")
    current_state["review"] = "不需要调整！"
    await ctx.set("state", current_state)
    return "完成评审。"

write_agent = FunctionAgent(
        name="WriteAgent",
        # 修改描述为中文
        description="用于撰写关于给定主题的报告。",
        # 修改系统提示词为中文
        system_prompt=(
            "你是撰写代理，能够撰写关于给定主题的报告。 "
            "你的报告应以Markdown格式编写，内容应基于研究笔记。 "
            "报告撰写完成后，你应该至少调用评审代理取一次反馈。"
        ),
        llm=llm,
        tools=[write_report],
        can_handoff_to=["ReviewAgent", "ResearchAgent"],
    )

    review_agent = FunctionAgent(
        name="ReviewAgent",
        # 修改描述为中文
        description="用于评审报告并提供反馈。",
        # 修改系统提示词为中文
        system_prompt=(
            "你是评审代理，能够评审报告并提供反馈。 "
            "你的反馈应该要么批准当前报告，要么要求撰写代理进行修改。"
        ),
        llm=llm,
        tools=[review_report],
        can_handoff_to=["WriteAgent"],
    )

定义代理后，我们现在可以直接实例化 MY 以创建一个多代理系统。我们给它一个代理数组，并使用定义最初应该控制哪个代理。我们还可以定义变量的初始值，正如我们之前看到的，它是一个可以被所有代理访问的字典

    async def run_agent(user_msg: str):
        """Run the agent."""
        agent_workflow = AgentWorkflow(
            agents=[research_agent, write_agent, review_agent],
            root_agent="ResearchAgent",
            initial_state={
                "research_notes": {},
                "report_content": "Not written yet.",
                "review": "Review required.",
            }
        )
        ctx = Context(agent_workflow)
        handler = agent_workflow.run(
            user_msg=user_msg,
            ctx=ctx,
        )
        async for evt in handler.stream_events():
            if isinstance(evt, AgentStream):
                print(evt.delta, end="", flush=True)
    import asyncio
    # 测试运行
    usr_msg = """
    写一个关于Python的报告。
    """
    asyncio.run(run_agent(usr_msg))

完整示例

# 使用llm 代理agents
from llama_index.core.agent.workflow import ReActAgent, FunctionAgent, AgentWorkflow
from llama_index.core.workflow import JsonPickleSerializer, JsonSerializer
from llama_index.core.workflow import Context
from llama_index.llms.ollama import Ollama
from llama_index.core.agent.workflow import AgentStream

llm = Ollama(
    model="modelscope.cn/Qwen/Qwen2.5-7B-Instruct-GGUF:q5_k_m",
    temperature=0.7,
    request_timeout=60.0,
)

# 定义一个函数，模拟搜索函数
async def search_web(query: str) -> dict[str, str]:
    """根据查询返回相关信息。"""
    print(f"调用search: {query}")
    return {
        "query": query,
        "result": """
        Python 是一种解释型、面向对象、动态数据类型的高级程序设计语言。
        Python 由 Guido van Rossum 于 1989 年底发明，第一个公开发行版发行于 1991 年。
        像 Perl 语言一样, Python 源代码同样遵循 GPL(GNU General Public License) 协议。
        官方宣布，2020 年 1 月 1 日， 停止 Python 2 的更新。
        Python 2.7 被确定为最后一个 Python 2.x 版本。
        """
    }
async def record_notes(ctx: Context, notes: str, notes_title: str) -> str:
    """Useful for recording notes on a given topic."""
    print(f"调用record_notes: {notes}, {notes_title}")
    current_state = await ctx.get("state")
    if "research_notes" not in current_state:
        current_state["research_notes"] = {}
    current_state["research_notes"][notes_title] = notes
    await ctx.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."""
    print(f"调用write_report: {report_content}")
    current_state = await ctx.get("state")
    current_state["report_content"] = report_content
    await ctx.set("state", current_state)
    return "Report written."

async def review_report(ctx: Context, review: str) -> str:
    """Useful for reviewing a report and providing feedback."""
    print(f"调用review_report: {review}")
    current_state = await ctx.get("state")
    current_state["review"] = "不需要调整！"
    await ctx.set("state", current_state)
    return "完成评审。"

def test_llm_chat():
    research_agent = FunctionAgent(
        name="ResearchAgent",
        # 修改描述为中文
        description="用于在网络上搜索给定主题的信息并记录相关笔记。",
        # 修改系统提示词为中文
        system_prompt=(
            "你是研究代理，能够在网络上搜索给定主题的信息并记录相关笔记。 "
            "一旦笔记记录完成且你感到满意，你应该将控制权移交给撰写代理，让其撰写关于该主题的报告。"
        ),
        llm=llm,
        tools=[search_web,record_notes],
        can_handoff_to=["WriteAgent"]
    )
    write_agent = FunctionAgent(
        name="WriteAgent",
        # 修改描述为中文
        description="用于撰写关于给定主题的报告。",
        # 修改系统提示词为中文
        system_prompt=(
            "你是撰写代理，能够撰写关于给定主题的报告。 "
            "你的报告应以Markdown格式编写，内容应基于研究笔记。 "
            "报告撰写完成后，你应该至少调用评审代理取一次反馈。"
        ),
        llm=llm,
        tools=[write_report],
        can_handoff_to=["ReviewAgent", "ResearchAgent"],
    )

    review_agent = FunctionAgent(
        name="ReviewAgent",
        # 修改描述为中文
        description="用于评审报告并提供反馈。",
        # 修改系统提示词为中文
        system_prompt=(
            "你是评审代理，能够评审报告并提供反馈。 "
            "你的反馈应该要么批准当前报告，要么要求撰写代理进行修改。"
        ),
        llm=llm,
        tools=[review_report],
        can_handoff_to=["WriteAgent"],
    )
    async def run_agent(user_msg: str):
        """Run the agent."""
        agent_workflow = AgentWorkflow(
            agents=[research_agent, write_agent, review_agent],
            root_agent="ResearchAgent",
            initial_state={
                "research_notes": {},
                "report_content": "Not written yet.",
                "review": "Review required.",
            }
        )
        ctx = Context(agent_workflow)
        handler = agent_workflow.run(
            user_msg=user_msg,
            ctx=ctx,
        )
        async for evt in handler.stream_events():
            if isinstance(evt, AgentStream):
                print(evt.delta, end="", flush=True)
    import asyncio
    # 测试运行
    usr_msg = """
    写一个关于Python的报告。
    """
    asyncio.run(run_agent(usr_msg))

if __name__ == "__main__":
    test_llm_chat()

查看输出日志：

调用search: Python programming language
调用record_notes: Python 是一种解释型、面向对象、动态数据类型的高级程序设计语言。Python 由 Guido van Rossum 于 1989 年底发明，第一个公开发行版发行于 1991 年。像 Perl 语言一样, Python 源代码同样遵循 GPL(GNU General Public License) 协议。官方宣布，2020 年 1 月 1 日， 停止 Python 2 的更新。Python 2.7 被确定为最后一个 Python 2.x 版本。, Python编程语言
调用write_report: # Python 编程语言

## 引言
Python 是一种解释型、面向对象的和动态数据类型的高级程序设计语言。

## 历史背景
- **发明者**: Guido van Rossum。
- **发明时间**: 1989 年底开始开发，首个公开发行版于 1991 年发布。
- **停止更新**: 官方宣布在2020年1月1日停止Python 2的更新，并确定Python 2.7为最后一个Python 2.x版本。

## 许可证
Python 源代码遵循 GPL (GNU General Public License) 协议。
调用review_report: # Python 编程语言

## 引言
Python 是一种解释型、面向对象的和动态数据类型的高级程序设计语言。

## 历史背景
- **发明者**: Guido van Rossum。
- **发明时间**: 1989 年底开始开发，首个公开发行版于 1991 年发布。
- **停止更新**: 官方宣布在2020年1月1日停止Python 2的更新，并确定Python 2.7为最后一个Python 2.x版本。

## 许可证
Python 源代码遵循 GPL (GNU General Public License) 协议。
报告已通过评审，可以批准当前的Python编程语言报告。

批准此报告。