实践：构建多阶段工作流及智能体协作

可以设计并勾勒出具有智能体协作的多阶段工作流的实现方案。通过这种方式，多个大型语言模型（LLM）智能体，每个智能体都具有专门职责，能够系统地协作以完成一项复杂任务。一个主要应用是根据用户定义的主题生成一份全面的报告。

场景：自动化报告生成流程

我们的目标是构建一个系统，它接收用户提供的主题，并生成一份结构良好、信息丰富的报告。此过程自然地分解为几个阶段，每个阶段都适合一个专用智能体。这种分工使每个智能体都能专注于其特定专长，从而带来更有效的整体成果。

该工作流将包含以下智能体：

1. 研究智能体：接收主题，查询外部知识源（例如，模拟网络搜索、文档数据库）以收集相关信息。
2. 分析智能体：处理来自研究智能体的原始信息，提取相关事实，识别模式，并组织重要的观点。
3. 起草智能体：接收分析智能体的结构化观点，并撰写报告的连贯初稿。
4. 评审智能体：检查草稿的清晰度、准确性、语法和完整性。它可以批准草稿，或将其连同具体的修改意见发送回起草智能体。

可选地，在报告被认定完成前，可以加入一个最终的人工评审步骤，这与本章前面讨论的“人在回路”原则相符。

定义工作流和智能体交互

这些智能体之间的协作需要有清晰定义的流程。我们可以将其想象成一个有向图，其中节点代表智能体（或任务），边代表信息或控制的流动。

该图展示了报告生成工作流。每个智能体执行一个不同的步骤，将其输出传递给序列中的下一个智能体。请注意评审智能体到起草智能体的反馈循环，这使得迭代完善成为可能。

智能体接口设计

为了使智能体有效协作，它们需要有清晰定义的接口。每个智能体应期望某种输入格式，并生成可预测的输出格式。JSON是构建这些消息的常用选择。

例如，研究智能体可能：

• 输入：{"topic": "Advancements in Renewable Energy Storage"}
• 输出：{"topic": "Advancements in Renewable Energy Storage", "raw_data": [{"source": "url1", "content": "..."}, {"source": "doc2", "content": "..."}], "summary": "初步发现概览。"}

随后，分析智能体将此输出作为其输入，并生成类似以下内容：

• 输入：研究智能体的输出。
• 输出：{"topic": "Advancements in Renewable Energy Storage", "structured_insights": [{"point": "Lithium-ion battery improvements...", "details": "..."}, {"point": "Solid-state battery research...", "details": "..."}], "proposed_outline": ["简介", "锂离子", "固态", "结论"]}

这种结构化数据交换对于工作流阶段间的顺畅衔接至关重要。

实现编排器

编排器组件负责管理整个工作流。在这种简单的顺序工作流中，编排器依次调用每个智能体，将一个智能体的输出作为输入传递给下一个。对于更复杂的情况，你可能需要实现一个状态机或使用基于图的执行引擎，如“状态驱动和基于图的编排模型”中所讨论的。

以下是简单编排器的高级Python类结构示例：

# 假设智能体类（ResearchAgent, AnalysisAgent等）在其他地方定义。
# 每个智能体类都有一个“run”方法，接收一个字典并返回一个字典。

class ReportOrchestrator:
    def __init__(self):
        self.research_agent = ResearchAgent()
        self.analysis_agent = AnalysisAgent()
        self.drafting_agent = DraftingAgent()
        self.review_agent = ReviewAgent()
        self.max_revisions = 2 # 防止无限循环

    def generate_report(self, topic: str) -> dict:
        print(f"开始为主题生成报告: {topic}")

        # 阶段 1: 研究
        research_input = {"topic": topic}
        research_output = self.research_agent.run(research_input)
        print("研究完成。")

        if not research_output.get("raw_data"):
            print("研究智能体未能找到数据。中止。")
            return {"status": "error", "message": "未找到数据。"}

        # 阶段 2: 分析
        analysis_output = self.analysis_agent.run(research_output)
        print("分析完成。")

        # 阶段 3 和 4: 起草与评审循环
        current_draft_input = analysis_output
        final_report = None

        for attempt in range(self.max_revisions + 1):
            print(f"起草尝试 {attempt + 1}...")
            draft_output = self.drafting_agent.run(current_draft_input)
            print("起草完成。")

            print("评审草稿中...")
            review_input = draft_output # 传递完整的草稿上下文
            review_output = self.review_agent.run(review_input)
            print("评审完成。")

            if review_output.get("status") == "approved":
                final_report = review_output.get("final_report", draft_output.get("draft_text"))
                print("报告已批准！")
                break
            elif attempt < self.max_revisions:
                print("请求修改。准备重新起草。")
                # review_output 应包含给起草智能体的反馈
                # 此反馈需要整合到下一次起草迭代的输入中
                current_draft_input = {
                    **analysis_output, # 原始观点
                    "previous_draft": draft_output.get("draft_text"),
                    "feedback": review_output.get("feedback")
                }
            else:
                print("已达到最大修改次数。报告未获批准。")
                final_report = review_output.get("final_report_as_is", draft_output.get("draft_text")) # 或者作为错误处理
                break

        # 阶段 5: 可选的人工评审（在此主流程之外，但已注明）
        if final_report:
            print("报告生成过程完成。")
            return {"status": "success", "report": final_report}
        else:
            print("报告生成在修改后失败。")
            return {"status": "error", "message": "未能生成一份获批准的报告。"}

# 示例用法：
# orchestrator = ReportOrchestrator()
# report_topic = "量子计算对密码学的影响"
# result = orchestrator.generate_report(report_topic)
# if result["status"] == "success":
#     print("\n最终报告:\n", result["report"])
# else:
#     print("\n报告生成失败:\n", result["message"])

在此伪代码中：

• 每个智能体都被实例化。
• generate_report 方法执行序列。
• 数据被显式传递。
• 在起草智能体和评审智能体之间实现了一个简单的修订循环。ReviewAgent 需要设计为输出状态（approved 或 needs_revision）和详细反馈。DraftingAgent 则需要能够接受此反馈以改进其下一稿。

智能体实现考量

虽然我们在此不详述每个智能体的完整大型语言模型（LLM）提示工程（因为这已在预备材料和智能体设计的前面章节中涵盖），但请考虑每个智能体 run 方法的以下几点：

• 研究智能体：其LLM提示将侧重于识别搜索查询，可能处理搜索结果（如果使用了工具），并总结发现。它还可能涉及决定何时收集到足够的信息。
• 分析智能体：该智能体的提示将指导LLM筛选研究者的输出，识别核心主题，丢弃不相关信息，并逻辑地组织剩余数据。它可能生成一个重要观点的列表或一个结构化的JSON对象。
• 起草智能体：该提示将指示LLM将结构化观点扩展为流畅的叙述，并遵循指定的语气或风格。它可能会从分析智能体接收一个大纲。
• 评审智能体：该智能体可以使用基于清单的提示策略。它将要求LLM根据连贯性、事实准确性（如果可对照提供的源片段验证）、语法正确性和基于初始结构化观点的完整性等标准评估草稿。其输出应明确说明草稿是否获批准或需要修订，并附带具体的、可操作的反馈。

运行与观察工作流

要真正构建此系统，您需要与您选择的LLM框架集成。在执行期间，您将监控：

• 智能体之间传递的数据对象。
• 每个智能体发出的LLM调用。
• 中间输出（例如，原始研究数据、结构化观点、草稿的每个版本）。
• 最终报告的质量。

日志记录（如第6章所述）在此处非常重要。全面的日志使您能够追踪数据和决策的流动，这对于调试和理解智能体行为不可或缺。

扩展工作流：进一步实践

一旦这个基本流程可用，可以将以下增强视为进一步练习，并借鉴本章中的其他理念：

1. 错误处理和韧性：如果智能体失败（例如，API错误、格式错误的输出），会发生什么？在编排器或单个智能体内部实现try-catch块。系统将如何恢复或报告失败？（与“处理故障和确保可靠性”相关）。
2. 自适应分支：修改编排器以根据智能体输出做出决策。例如，如果研究智能体发现的信息很少，编排器可能会在进行分析之前触发一个“主题细化智能体”。（与“自适应任务规划”相关）。
3. 并行执行：此工作流的哪些部分可以并行化？例如，如果主题有多个子方面，多个研究智能体能否并行工作，并在分析前聚合它们的结果？（与“管理资源和智能体工作负载分配”相关）。
4. 工作流中期决策中的“人在回路”：引入一个步骤，在起草开始前，由人工评审分析智能体产生的结构化观点，从而允许纠正方向。（与“纳入人工监督”相关）。
5. 工具集成：为研究智能体配备真正的网络搜索工具或文档检索能力。这涉及处理API调用和管理外部工具返回的数据。（与第2章的“为智能体集成外部工具和函数”相关）。

核心要点

构建多阶段工作流的本次练习应巩固您对以下方面的理解：

• 清晰角色定义的必要性：每个智能体都需要有明确的目的和范围。
• 结构化智能体间通信的重要性：标准化数据格式对于顺畅的交接至关重要。
• 编排器的作用：即使是简单的编排器也能提供对任务序列和流程的基本控制。
• 迭代开发：工作流通常涉及循环和完善，例如草稿-评审循环。
• 可扩展性和复杂性：这个相对简单的例子提供了一个基础，在此之上可以构建涉及更多智能体和更复杂依赖关系的更复杂协作。

通过实践此类系统的设计和（甚至是高层级的）实现，您已在掌握复杂多智能体LLM应用构建的道路上前进。此处遇到的模式和挑战代表了更先进和更大规模智能体团队中出现的那些情况。