在上一篇文章中
我们探讨了多代理系统的概念，以及 Clearwater Analytics 如何通过 CWIC Flow 开创了这一方法。我们还展示了如何在不使用专门框架的情况下构建一个简单的多代理系统。现在，让我们来探索 LangGraph，这是一个功能强大的库，为构建复杂多代理系统提供结构化工作流。
LangGraph 是基于 LangChain 构建的库，允许您创建具有状态的多代理应用程序，使用大语言模型（LLMs）。它提供了一种结构化的方式，将链组合成图，管理多次 LLM 调用之间的状态，并构建具有条件路由的复杂工作流。
了解 LangGraph 的基础知识
在深入实现之前，让我们先了解 LangGraph 中的一些关键概念：
•状态（State）：在图的不同步骤之间持久化的信息•节点（Nodes）：处理状态并返回新状态的函数•边（Edges）：定义状态在节点之间的流动•图（Graph）：通过边连接节点形成整体工作流
这种结构化方法允许代理之间进行更复杂的交互，包括条件路由、反馈循环和持久化记忆。
LangGraph 工作流可视化
让我们可视化一个基本的 LangGraph 如何工作：
此图表说明了关键概念：
•状态（蓝色框）：包含流经图的数据•节点（紫色框）：处理状态并生成新状态的函数•边（箭头）：定义从一个节点到下一个节点的流动•入口/出口点（绿色椭圆）：工作流的起点和终点
环境设置
首先，确保安装所有必要的软件包。如果您是第一次运行此代码，需要安装以下软件包。
安装必要的软件包
# 安装最新版本的必要软件包!pip install langchain langchain-openai langgraph python-dotenv
现在，设置环境变量。我们需要一个 OpenAI 的 API 密钥来使用其模型。在与此笔记本相同的目录下创建一个 .env 文件，内容如下：
OPENAI_API_KEY=您的_api_密钥
# 从 .env 文件加载环境变量import osfrom dotenv import load_dotenvload_dotenv()# 从 .env 文件加载 API 密钥# 验证 API 密钥是否加载if os.getenv("OPENAI_API_KEY")isNone:print("警告：未在环境变量中找到 OPENAI_API_KEY。")else:print("已在环境变量中找到 OPENAI_API_KEY。")导入必要的库
现在，导入我们构建多代理系统所需的库：
# 导入必要的库from typing importDict,List,TypedDict,Annotated,Sequence,Any,Optional,Literal,Unionimport json# 用于 langchain 和 langgraph 的现代导入from langchain_core.messages importHumanMessage,AIMessage,SystemMessage,BaseMessagefrom langchain_openai importChatOpenAIfrom langgraph.graph importStateGraph,END
创建一个简单的 LLM 辅助函数
在深入多代理系统之前，我们先创建一个简单的辅助函数与 LLM 交互。这将帮助我们理解 LangGraph 的基本构建模块。
def ask_llm(prompt, model="gpt-4o", temperature=0.7):"""一个简单的函数，用于从 LLM 获取响应。"""# 创建 ChatOpenAI 实例llm =ChatOpenAI(model=model, temperature=temperature)# 使用提示创建消息messages =[HumanMessage(content=prompt)]# 从 LLM 获取响应response = llm.invoke(messages)# 返回响应的内容return response.content# 测试我们的函数response = ask_llm("LangGraph 是什么，它与 LangChain 有何关系？")print(response)
输出响应：
LangGraph 是一个开源库，设计用于构建和管理使用大语言模型（LLMs）的复杂应用程序。它是一个框架，提供工具和组件来创建、定制和执行涉及 LLM 的工作流，方便将这些模型集成到各种应用中。LangGraph 与 LangChain 密切相关，可视为 LangChain 的扩展或补充工具集。LangChain 是一个专门为开发 LLM 应用程序设计的框架，专注于将不同组件或任务链接起来以创建更复杂的工作流。LangGraph 在 LangChain 的基础上提供了额外的功能，以更结构化的方式定义和管理基于 LLM 的应用程序。本质上，LangChain 专注于语言模型驱动任务的链接，而 LangGraph 提供了更全面的框架，包括基于图的执行流、对任务的增强控制以及对不同组件交互的改进管理。两者结合可用于简化利用 LLM 功能的复杂应用程序的开发和部署。
创建一个简单的图
让我们创建一个只有一个代理的简单图，以理解这些概念：
# 定义状态类型classSimpleState(TypedDict):messages:List[BaseMessage]# 对话历史next: str # 图中的下一步# 定义一个简单的代理节点def simple_Agent(state:SimpleState)->SimpleState:"""一个简单的代理，响应最后一条消息。"""messages = state["messages"]llm =ChatOpenAI(model="gpt-4o", temperature=0.7)response = llm.invoke(messages)return{"messages": messages +[response],"next":"output"}# 定义输出节点def output(state:SimpleState):"""返回最终状态，标记工作流结束。"""return{"messages": state["messages"],"next":END}# 构建图def build_simple_graph():"""构建一个简单的 LangGraph 工作流。"""# 创建一个新图workflow =StateGraph(SimpleState)# 添加节点workflow.add_node("agent", simple_agent)workflow.add_node("output", output)# 添加边workflow.add_edge("agent","output")# 设置入口点workflow.set_entry_point("agent")# 编译图return workflow.compile()
现在，运行我们的简单图：
# 创建图simple_graph = build_simple_graph()# 使用测试消息初始化状态initial_state ={"messages":[HumanMessage(content="用简单术语解释什么是多代理系统。")],"next":""}# 运行图result = simple_graph.invoke(initial_state)# 打印对话for message in result["messages"]:if isinstance(message,HumanMessage):print(f"用户: {message.content}")elif isinstance(message,AIMessage):print(f"AI: {message.content}")输出：
用户: 用简单术语解释什么是多代理系统。
AI: 多代理系统是一组相互交互的独立“代理”。可以将每个代理想象成一个能够独立思考和行动的个体，类似于人类。这些代理一起工作或竞争以实现特定目标或完成任务。
简单来说，想象一群机器人清理你的房子。每个机器人是一个代理。一个可能负责吸尘地板，另一个可能负责擦拭家具，还有一个可能负责倒垃圾。它们各自独立工作，但通过沟通和协调确保整个房子被高效清洁。这就是多代理系统的本质：多个实体以协调的方式一起工作，以执行复杂任务或解决问题。
此图表显示：
•从开始 → 代理 → 输出 → 结束的整体流程•代理处理子图，显示 simple_agent 函数的内部步骤•状态子图，显示 SimpleState TypedDict 的组件•使用颜色编码的节点区分不同类型的组件
工作流从包含消息和 next 字段的初始状态开始，通过代理节点处理，代理节点使用 LLM 添加响应，然后传递到输出节点，返回最终状态，结束工作流。
接下来发生了什么
我们定义了状态模式（SimpleState），包含我们想要跟踪的字段。
我们为代理和输出函数创建了节点。
我们通过边连接这些节点，定义信息的流动。
我们设置了入口点并编译了图。
我们用一条用户消息初始化状态并运行了图。
这个简单示例说明了 LangGraph 的基本概念。在现实世界的场景中，您可以创建具有多个节点和条件边的更复杂图。
下一步
下一步，我们将构建我们的第一个专门代理：研究者代理。我们将深入探讨代理设计模式，学习如何定义代理角色和能力，并了解如何使代理有效地执行研究任务。
创建专门的研究者代理
在之前的步骤中，我们设置了环境并创建了一个只有一个代理的简单图。现在，我们将为协作研究助手系统构建第一个专门代理：研究者代理。
在本步骤中，我们将专注于构建研究者代理——负责收集和分析信息的组件。
什么是研究者代理？
研究者代理是一个人工智能系统，设计用于深入调查主题并提供全面、结构化的信息。它是我们多代理系统中的信息收集组件，具有以下能力：
•深入理解研究查询•全面收集相关信息•以结构化、可访问的格式组织发现•客观、准确地呈现信息•识别可用信息中的局限性和空白
使用 LangGraph 构建现代研究者代理
LangGraph 和 LangChain 的最新版本提供了改进的方法来构建代理。让我们探索如何使用这些现代工具创建研究者代理。
步骤 1：设置环境
首先，我们需要导入必要的库。请注意 LangGraph 和 LangChain 的现代导入路径：
from typing importDict,List,TypedDict,Any,Optionalimport jsonimport osfrom dotenv import load_dotenv# 用于 langchain 和 langgraph 的现代导入from langchain_core.messages importHumanMessage,AIMessage,SystemMessage,BaseMessagefrom langchain_openai importChatOpenAIfrom langgraph.graph importStateGraph,END# 加载环境变量load_dotenv()
步骤 2：定义研究者的角色
清晰的系统提示对于塑造代理的行为至关重要：
RESEARCHER_SYSTEM_PROMPT ="""您是一个熟练的研究者代理，任务是收集有关给定主题的全面信息。您的职责包括：1.分析研究查询，了解所需信息2.进行彻底研究，收集相关事实、数据和观点3.以清晰、结构化的格式组织信息4.确保发现的准确性和客观性5.引用来源或指出可能需要验证的信息6.识别信息中的潜在空白以清晰的结构化格式呈现您的发现，适当使用章节和项目符号。您的目标是提供全面、准确且有用的信息，充分回答研究查询。"""
步骤 3：创建研究者函数
现在我们将研究者代理实现为一个函数：
def create_researcher_agent(model="gpt-4o", temperature=0.7):"""使用指定的 LLM 创建研究者代理。"""# 初始化模型 llm =ChatOpenAI(model=model, temperature=temperature)def researcher_function(messages):"""处理消息并返回研究者代理响应的函数。"""# 如果消息为空或第一条消息不是系统提示，则添加系统提示ifnot messages ornot isinstance(messages[0],SystemMessage)or messages[0].content != RESEARCHER_SYSTEM_PROMPT: messages =[SystemMessage(content=RESEARCHER_SYSTEM_PROMPT)]+(messages if isinstance(messages, list)else[])# 从 LLM 获取响应 response = llm.invoke(messages)return responsereturn researcher_function
使用 LangGraph 集成：现代方法
LangGraph 的最新版本使用 TypedDict 进行状态管理，提供更好的类型安全性和更清晰的状态结构：
# 为研究工作流定义状态类型classResearchState(TypedDict):"""我们研究工作流状态的类型定义。""" messages:List[BaseMessage]# 对话历史 query: str # 研究查询 research:Optional[str]# 研究发现next:Optional[str]# 图中的下一步
创建研究者节点
接下来，我们为 LangGraph 工作流实现一个节点：
def researcher_node(state:ResearchState)->ResearchState:"""在图中执行研究查询的节点。"""# 从状态中获取查询 query = state["query"]# 为研究者创建特定消息 research_message =HumanMessage(content=f"请彻底研究以下主题：{query}")# 获取研究者代理 researcher = create_researcher_agent()# 从研究者代理获取响应 response = researcher([research_message])# 使用研究发现更新状态 new_messages = state["messages"]+[research_message, response]# 返回更新后的状态return{**state,"messages": new_messages,"research": response.content,"next":"output"# 在多代理系统中，这将转到下一个代理}构建研究工作流
现在我们可以构建一个完整的 LangGraph 工作流：
def build_research_graph():"""使用 LangGraph 构建一个简单的研究工作流。"""# 使用我们的状态类型创建一个新图 workflow =StateGraph(ResearchState)# 添加节点 workflow.add_node("researcher", researcher_node) workflow.add_node("output", output_node)# 添加边 workflow.add_edge("researcher","output")# 设置入口点 workflow.set_entry_point("researcher")# 编译图return workflow.compile()
使用结构化输出增强研究者代理
为了更好地与其他代理集成，我们可以增强研究者代理以提供结构化输出：
ENHANCED_RESEARCHER_PROMPT ="""您是一个熟练的研究者代理，任务是收集有关给定主题的全面信息。您的职责包括：1.分析研究查询，了解所需信息2.进行彻底研究，收集相关事实、数据和观点3.以清晰、结构化的格式组织信息4.确保发现的准确性和客观性5.引用来源或指出可能需要验证的信息6.识别信息中的潜在空白以以下结构化格式呈现您的发现：SUMMARY：发现的简要概述（2-3句）KEY POINTS：-重点1-重点2-重点3DETAILED FINDINGS：1.[主题领域1]-详细说明和解释-支持证据-适用的不同观点2.[主题领域2]-详细说明和解释-支持证据-适用的不同观点GAPS AND LIMITATIONS：-识别信息可能不完整的领域-指出任何矛盾或争议领域-建议可能需要的额外研究您的目标是提供全面、准确且有用的信息，充分回答研究查询。"""
更强大的研究工作流
使用我们增强的研究者代理，我们可以创建一个更强大的工作流：
我们使用专门的状态类型实现此工作流：
classEnhancedResearchState(TypedDict):"""我们研究工作流状态的增强类型定义。""" messages:List[BaseMessage] query: str structured_research:Optional[str]next:Optional[str]
研究者代理实战：示例查询
让我们看看我们的增强研究者代理如何处理复杂查询：
查询：“可再生能源技术的最新进展是什么？”
响应（结构化格式）：
SUMMARY：可再生能源技术的最新进展包括太阳能电池效率的显著提高、创新的风力涡轮机设计、增强的能源存储解决方案、绿色氢气生产的突破以及高级生物燃料的发展。
KEY POINTS：
•太阳能光伏技术通过钙钛矿-硅串联电池实现了超过 30% 的创纪录效率水平•浮式海上风力涡轮机正在扩展深水部署的可能性•电网规模电池存储成本大幅下降，同时能量密度增加•通过电解法生产的绿色氢气变得更高效且经济上可行•来自非食品生物质的高级生物燃料正在接近商业化可行性
DETAILED FINDINGS：
1.太阳能进展•钙钛矿太阳能电池迅速改进，实验室效率超过 25%•钙钛矿-硅串联电池达到 31.25% 的效率，接近理论极限•双面太阳能板（捕获两侧光线）将能量产出提高 5–30%•建筑一体化光伏（BIPV）通过透明太阳能电池用于窗户取得了进展•农业光伏系统（结合农业与太阳能生产）显示出令人期待的结果2.风能创新•叶片长度超过 100 米的大型涡轮机提高了容量因子•浮式海上风力平台可在 60 米以上深度的水域部署•空中风能系统（基于风筝和无人机的系统）处于高级测试阶段•无叶片风力涡轮机通过振荡技术减少野生动物影响和维护成本•数字孪生技术和人工智能改善预测性维护和输出优化
（为简洁起见，省略了其他部分）
GAPS AND LIMITATIONS：
•长时能源存储（10+小时）尽管有所进展仍是一个重大挑战•提到的突破性技术大多仍处于从实验室到商业部署的扩展阶段•新兴技术之间的成本比较常常缺乏标准化指标•某些新技术（如某些稀土元素开采用于磁铁）的环境影响需要进一步研究•技术采用和政策支持的地区差异导致进展不均衡
现代方法的优势
使用 LangGraph 构建研究者代理的现代方法提供了以下优势：
•类型安全：TypedDict 提供更好的类型检查和文档•结构化状态：清晰定义的状态结构提高了可维护性•模块化设计：函数和组件可以轻松重用和测试•现代 API：使用最新的 LangChain 和 LangGraph API 确保兼容性
下一步：构建完整的多代理系统
研究者代理只是构建全面多代理系统的第一步。在未来的文章中，我们将探索：
•构建批评者代理以评估和挑战研究者的发现•创建编写者代理以将信息合成为连贯的内容•实现协调者代理以管理代理之间的工作流•开发高级路由逻辑以实现更动态的代理交互
使用 LangGraph 构建现代研究者代理为多代理系统提供了强大的基础。通过利用最新的 API 和结构化方法进行代理开发，我们可以创建更有效、可维护且强大的 AI 系统，协作解决复杂问题。
我们构建的研究者代理展示了如何设计专门的代理，在更大系统中履行特定角色——收集和组织信息，随后由工作流中的其他代理进行评估、完善和呈现。
随着我们继续探索多代理系统，我们将看到这些专门组件如何协作，创建超越单一模型能力的 AI 解决方案。
添加批评者代理
在之前的步骤中，我们设置了环境并创建了一个研究者代理。现在，我们将为协作研究助手系统添加第二个专门代理：批评者代理。
批评者代理的概念
批评者代理旨在评估、挑战和改进其他代理的工作。正如人类工作受益于同行评审和建设性批评一样，人工智能输出也可以通过专门的评估得到显著提升。
批评者代理在多代理工作流中履行以下关键功能：
•质量保证：识别研究中的不准确、空白或缺陷•偏见检测：指出信息中的潜在偏见•完整性检查：确保主题的各个方面得到充分覆盖•替代观点：引入不同的观点或解释•建设性反馈：提供可操作的改进建议
使用 LangGraph 进行现代实现
LangGraph 和 LangChain 的最新版本为构建更健壮和可维护的多代理系统提供了强大的工具。让我们探索如何使用这些现代框架实现批评者代理。
核心组件
我们的实现专注于三个关键元素：
•类型安全：使用 TypedDict 进行状态管理•结构化输出：使用结构化 JSON 反馈增强批评者•模块化设计：创建可重用的代理函数
批评者的系统提示
批评者代理的基础是一个精心设计的系统提示，确立其角色和职责：
CRITIC_SYSTEM_PROMPT ="""您是批评者代理，属于协作研究助手系统的一部分。您的角色是评估和挑战研究者代理提供的信息，以确保其准确性、完整性和客观性。您的职责包括：1.分析研究发现的准确性、完整性和潜在偏见2.识别信息中的空白或逻辑不一致3.提出可能被忽略的重要问题4.建议改进或替代观点5.确保最终信息平衡且全面您的批评应具有建设性，目标不是否定研究者的工作，而是增强其质量。以清晰、有组织的方式格式化您的反馈，突出需要注意的具体点。请记住，您的最终目标是确保最终研究输出的质量尽可能高。"""
现代状态管理
LangGraph 最新版本的重大改进是通过 TypedDict 进行更好的状态管理：
classCollaborativeResearchState(TypedDict):"""协作研究助手的状态类型。""" messages:List[BaseMessage]# 对话历史next:Optional[str]# 图中的下一步
这种方法提供了清晰的类型提示，使代码更易于维护。
工作流：研究者 → 批评者
让我们可视化研究者和批评者代理之间的基本信息流：
我们的 LangGraph 实现使用现代语法定义此工作流：
def build_collaborative_research_assistant():"""构建一个具有研究者和批评者代理的协作研究助手。"""# 使用我们的状态类型创建一个新图 workflow =StateGraph(CollaborativeResearchState)# 添加节点 workflow.add_node("researcher", researcher_node) workflow.add_node("critic", critic_node) workflow.add_node("output", output_node)# 添加边 workflow.add_edge("researcher","critic") workflow.add_edge("critic","output")# 设置入口点 workflow.set_entry_point("researcher")# 编译图return workflow.compile()使用结构化输出增强批评者
为了使批评者代理对下游流程更有用，我们可以增强其以 JSON 格式提供结构化反馈。这便于其他代理（如编写者代理）处理和整合批评。
以下是我们实现结构化批评的方式：
classCriticEvaluation(BaseModel):"""批评者评估的结构化格式。""" quality_score:int=Field(description="总体质量评分，1-10") strengths:List[str]=Field(description="研究的关键优势") areas_for_improvement:List[str]=Field(description="需要改进的领域") missing_information:List[str]=Field(description="未包含的重要信息") bias_assessment: str =Field(description="研究中潜在偏见的评估") additional_questions:List[str]=Field(description="应解决的问题")
这种结构化方法使我们能够构建更健壮的工作流：
完整的多代理研究流程
将所有部分组合在一起，我们的多代理系统遵循以下流程：
1.用户输入：用户提交研究问题2.研究阶段：研究者代理收集和组织信息3.批评阶段：批评者代理评估研究的质量和完整性4.结构化反馈：系统提供有组织、可操作的批评5.最终输出：结合研究和批评提供全面的响应
现代方法的优势
使用最新的 LangGraph 和 LangChain API 的现代实现提供了以下优势：
•类型安全：正确类型的状态减少错误并提高代码可维护性•结构化输出：JSON 格式的批评便于与其他组件集成•更好的可维护性：通过模块化代理设计实现清晰的关注点分离•可视化支持：内置工具支持工作流可视化和理解
通过向多代理系统添加批评者代理，我们创建了一个强大的检查与平衡机制，提高了人工智能生成研究的质量、准确性和完整性。这种方法模仿了人类的协作过程，其中同行评审和建设性批评带来更好的结果。
结构化的批评方法还为多代理旅程的下一步奠定了基础：添加一个编写者代理，将研究和批评合成为连贯、精心制作的最终响应。
在下一部分中，我们将探索如何构建这个编写者代理并完成我们的多代理研究助手系统。
编写者代理
在本文的前几部分中，我们探讨了如何通过专门的代理构建协作研究助手：研究者代理收集信息，批评者代理评估和挑战这些信息。现在，是时候添加第三个关键组件：编写者代理，它将这些信息合成为连贯、精心撰写的响应。本节探讨如何实现一个编写者代理，合成信息并生成连贯、全面的输出——创建一个完整的协作研究工作流，模仿人类研究者、编辑者和作者团队。
编写者代理的概念
编写者代理在多代理系统中作为最终的沟通者，将原始研究和批评分析转化为适合人类消费的精炼、连贯内容。
编写者代理在工作流中履行以下关键功能：
•信息合成：结合研究者的发现和批评者的评估中的见解•组织：以逻辑、易于访问的方式为最终用户结构化信息•风格适应：以清晰、引人入胜的语言呈现信息，适合上下文•观点整合：平衡不同观点，创建全面的响应•清晰度增强：简化复杂概念而不牺牲准确性
使用 LangGraph 进行现代实现
LangGraph 和 LangChain 的最新版本为构建集成多代理系统提供了复杂的工具。让我们探索如何使用这些现代框架实现编写者代理。
编写者的系统提示
编写者代理的基础是一个精心设计的系统提示：
WRITER_SYSTEM_PROMPT ="""您是编写者代理，属于协作研究助手系统的一部分。您的角色是合成研究者代理提供的信息和批评者代理的反馈，生成连贯、精心撰写的响应。您的职责包括：1.分析研究者提供的信息和批评者的反馈2.以逻辑、易于理解的结构组织信息3.以清晰、引人入胜的写作风格呈现信息4.平衡不同观点并确保客观性5.创建全面、准确且精心撰写的最终响应以清晰、有组织的方式格式化您的响应，适当使用标题、段落和项目符号。使用简单语言解释复杂概念，并在有帮助时提供示例。请记住，您的目标是创建有效向用户传达信息的最终响应。"""
现代状态管理和节点实现
LangGraph 最新版本的一个关键改进是使用 TypedDict 进行状态管理，提供更好的代码组织和类型安全：
classCollaborativeResearchState(TypedDict):"""协作研究助手的状态类型。""" messages:List[BaseMessage]# 对话历史next:Optional[str]# 图中的下一步
编写者节点的实现遵循简洁的功能方法：
def writer_node(state:CollaborativeResearchState)->CollaborativeResearchState:"""编写者代理的节点函数。"""# 从状态中提取消息 messages = state["messages"]# 使用系统提示创建编写者消息 writer_messages =[SystemMessage(content=WRITER_SYSTEM_PROMPT)]+ messages# 初始化 LLM，平衡创造力和准确性 llm =ChatOpenAI(model="gpt-4o", temperature=0.6)# 获取编写者的响应 response = llm.invoke(writer_messages)# 返回更新后的状态return{"messages": messages +[response],"next":"output"}完整的多代理工作流
当我们将编写者代理与研究者和批评者代理集成时，我们创建了一个复杂的工作流，模仿专业的科研和写作团队：
此工作流的 LangGraph 实现简洁且可维护：
def build_complete_research_assistant():"""构建一个具有研究者、批评者和编写者代理的完整研究助手。"""# 使用我们的状态类型创建一个新图 workflow =StateGraph(CollaborativeResearchState)# 添加节点 workflow.add_node("researcher", researcher_node) workflow.add_node("critic", critic_node) workflow.add_node("writer", writer_node) workflow.add_node("output", output_node)# 添加边 workflow.add_edge("researcher","critic") workflow.add_edge("critic","writer") workflow.add_edge("writer","output")# 设置入口点 workflow.set_entry_point("researcher")# 编译图return workflow.compile()高级应用的增强功能
对于更复杂的应用，我们可以通过以下高级功能增强编写者代理和整体工作流：
丰富元数据跟踪
我们可以创建一个增强的状态类型，跟踪每个代理贡献的元数据：
classEnhancedResearchState(TypedDict):"""研究过程元数据的增强状态类型。""" messages:List[BaseMessage]# 对话历史 metadata:Dict[str,Any]# 每个步骤的元数据next:Optional[str]# 图中的下一步
这允许我们捕获有关处理时间、令牌计数和模型参数的信息，这对于分析和优化非常有价值：
灵活的架构模式
我们构建的三代理结构（研究者 → 批评者 → 编写者）可以适应各种应用需求：
1.层级组织：•研究团队（多个专门研究者）→ 编辑者 → 编写者2.并行处理：•多个研究者处理不同方面 → 批评者评估每个部分 → 编写者合成所有内容3.混合结构：•部分代理按顺序工作，部分并行，基于任务需求
编写者代理的优势
将编写者代理添加到多代理系统提供以下关键优势：
•改进沟通：内容以优化人类理解的方式呈现•一致性：即使研究涵盖不同主题，写作风格和结构保持一致•整合：各种观点和信息片段被连贯地编织在一起•效率：最终输出无需额外编辑或重新格式化即可使用•适应性：编写者代理可根据上下文和受众调整风格和结构
通过向系统添加编写者代理，我们完成了一个强大的协作工作流，模仿人类在复杂信息任务上的协作方式。系统中每个代理都有专门的角色：
•研究者代理收集全面信息•批评者代理评估并确保质量•编写者代理有效地向最终用户传达信息
这种关注点分离使每个代理在其特定任务上表现出色，同时为整体贡献。使用 LangGraph 和 LangChain 的现代功能，如 TypedDict 进行状态管理，我们构建了一个不仅强大而且可维护和可扩展的系统。
多代理系统代表了 AI 应用设计的重要进步，超越了单一代理方法的局限性，创建了更健壮、平衡和有效的解决方案。
协调者代理
在探索使用 LangGraph 构建复杂多代理系统的过程中，我们已经探讨了研究、批评和编写等专门代理。现在，让我们聚焦于将固定、顺序工作流转变为真正动态、智能解决方案的关键组件：协调者代理。
多代理系统中协调的需要
随着多代理系统复杂性的增加，协调代理之间的交互变得越来越重要。一个设计良好的协调者可以：
•分析用户查询以确定最佳工作流•对于简单请求跳过不必要的步骤•在需要时启动额外的研究循环•确保所有相关专家适当贡献•根据中间结果调整工作流
协调者代理架构
与之前的代理按固定顺序操作的实现不同，此系统使用协调者代理在每个阶段确定最佳路径。对于简单查询，它可以完全绕过研究，而对于复杂查询，它协调完整的科研循环。
在现代 LangGraph 中构建协调者
现代 LangGraph 为实现这种动态架构提供了强大的工具：
def coordinator_node(state:ResearchState)->ResearchState:"""决定工作流路径的协调者节点。"""# 从状态中提取消息 messages = state["messages"]# 使用系统提示创建协调者消息 coordinator_messages =[SystemMessage(content=COORDINATOR_SYSTEM_PROMPT)]+ messages# 初始化 LLM，使用较低温度以确保一致的决策 llm =ChatOpenAI(model="gpt-4o", temperature=0.2)# 获取协调者的响应 response = llm.invoke(coordinator_messages)# 解析 JSON 响应以确定下一步try: decision = json.loads(response.content) next_step = decision.get("next","researcher")# 如果未指定，默认为研究者exceptException:# 如果解析 JSON 出错，默认为研究者 next_step ="researcher"# 返回更新后的状态return{"messages": messages,"next": next_step}我们实现的关键改进是工作流图中使用条件边：
# 从协调者添加条件边workflow.add_conditional_edges("coordinator",lambda state: state["next"],{"researcher":"researcher","done":"output"})动态决策实战
让我们看看协调者代理的决策过程如何工作：
对于简单的事实查询，协调者可以直接提供答案，避免不必要的工作。对于需要专业知识的复杂主题，它协调涉及所有专门代理的完整研究循环。
静态与动态工作流的比较
与静态工作流相比，带有协调者代理的动态方法提供：
•效率：简单查询获得即时响应•适应性：根据查询复杂性更改工作流•智能：在关键节点进行决策•递归：在需要时可循环进行进一步研究
技术实现细节
我们的实现利用了 LangGraph 的几个现代功能：
•TypedDict 进行状态管理：使用 ResearchState 进行适当的类型定义•条件边：基于代理决策的动态路由•结构化输出：带有推理和下一步的 JSON 格式•错误处理：具有回退的健壮解析•LLM 集成：在示例中使用 GPT-4o
这种方法提供了干净、可维护且可扩展的架构，可适应各种应用。
扩展协调者模式
协调者模式可以通过以下方式扩展：
构建动态多代理系统
通过实现协调者代理，我们可以将系统从固定、顺序工作流转变为动态系统，协调者决定采取哪条路径：
def build_dynamic_research_assistant():"""构建一个动态研究助手，协调者代理管理工作流。"""# 创建一个新图workflow =Graph()# 添加节点workflow.add_node("coordinator", coordinator_agent)workflow.add_node("researcher", researcher_agent)workflow.add_node("critic", critic_agent)workflow.add_node("writer", writer_agent)workflow.add_node("output", output)# 从协调者添加条件边workflow.add_conditional_edges("coordinator",lambda state: state["next"],{"researcher":"researcher","done":"output"})# 添加其余边workflow.add_edge("researcher","critic")workflow.add_edge("critic","writer")workflow.add_edge("writer","coordinator")workflow.add_edge("output",END)# 设置入口点workflow.set_entry_point("coordinator")# 编译图return workflow.compile()
此图创建了一个动态工作流，其中：
•协调者代理首先分析用户查询，决定是调用研究团队还是直接提供响应•如果需要研究，查询通过专门代理流转：研究者 → 批评者 → 编写者•编写者完成工作后，控制权返回协调者，协调者可以发送响应或启动额外研究•这创建了一个循环，复杂查询可以通过多次研究、批评和编写迭代进行优化
动态多代理系统架构的可视化
以下是我们动态多代理系统的可视化表示：
此图表说明了系统的动态性质：
•协调者代理首先接收所有用户查询•对于简单查询，协调者可以直接提供响应•对于复杂查询，协调者将其路由到专门代理•编写者代理生成响应后，协调者进行审查•协调者可以交付最终响应或请求额外研究
这创建了一个可以处理复杂、多步骤查询的反馈循环。
不同查询类型的决策流程
让我们也可视化不同类型的查询如何流经系统：
此图表说明了查询可以采取的不同路径：
•简单、事实查询获得直接响应•复杂研究查询通过专门代理•某些查询可能需要多次研究迭代
动态系统实战
让我们看看我们的动态系统如何处理不同类型的查询：
查询 1：“法国的首都是什么？”
协调者：
{"reasoning":"这是一个简单的实事问题，询问法国的首都。答案广为人知，无需深入研究、批评分析或专门编写。","next":"done"}系统响应：“法国的首都是巴黎。”
查询 2：“量子计算对网络安全的意义是什么？”
协调者：
{"reasoning":"此查询询问有关量子计算及其与网络安全关系的复杂信息。需要收集详细信息、评估不同观点并合成全面的响应。","next":"researcher"}[系统随后通过研究者、批评者和编写者代理处理此查询，然后返回协调者…]
编写者：[生成有关量子计算对加密、安全协议等影响的全面、结构化内容]
协调者（审查编写者的输出后）：
{"reasoning":"查询已通过多角度彻底研究、批评并合成为全面的响应，解决了量子计算对网络安全的意义。","next":"done"}系统响应：[将编写者的全面响应交付给用户]
协调者代理的价值
添加协调者代理以以下几种重要方式转变了我们的系统：
•效率：简单查询可以绕过不必要的处理，提供更快响应•适应性：系统根据每个查询的性质采取不同路径•迭代：复杂查询可以经过多次研究和优化•智能：系统在如何处理信息方面做出高层次决策•可扩展性：新专门代理可以更容易地添加到系统中
最重要的是，协调者使我们的系统感觉更像与一群智能专家交互，而不是固定的机械过程。
高级协调者实现
虽然我们的实现已经很强大，但有几种方法可以使协调者更加复杂：
•更多路由选项：允许协调者直接路由到任何代理，而不仅是研究者•反馈循环：使协调者能够识别空白并请求特定的额外研究•记忆管理：让协调者维护迄今学到的内容的摘要•元学习：允许协调者学习哪些路径对不同查询类型最有效•多查询规划：使协调者能够将复杂查询分解为子问题
协调者代理完成了我们的多代理研究助手，将其从顺序工作流转变为动态、智能系统。通过协调我们构建的专门代理——研究者、批评者和编写者——协调者创建了一个大于各部分之和的解决方案。
这种方法模仿了有效的人类团队工作方式：专门的专家在深思熟虑的协调下协作，解决复杂问题。结果是一个系统，可以快速提供简单答案，彻底处理复杂问题，并适应每种独特情况。
使用 LangGraph 的优势
使用 LangGraph 构建多代理系统提供了以下优势：
•结构化工作流：LangGraph 为定义代理交互提供清晰的结构，使复杂系统更易于设计和维护。•状态管理：框架处理多次 LLM 调用之间的状态管理，确保信息在代理之间正确流动。•条件路由：LangGraph 允许动态决定下一个代理，创建更适应和响应的系统。•检查点：内置检查点功能便于调试和恢复长时间运行的进程。•可扩展性：基于图的方法便于添加新代理或修改现有工作流，随着系统的发展。
与 CWIC Flow 的比较
虽然 LangGraph 为构建多代理系统提供了出色的框架，但值得注意的是它与 Clearwater 的 CWIC Flow 的相似之处和区别：
相似之处：
•两者都使用基于图的方法协调多个专门代理•两者都在代理交互中维护状态•两者都支持基于代理输出的条件路由
区别：
•CWIC Flow 专为金融服务设计，具有特定领域的代理和工具•CWIC Flow 包含 LangGraph 中没有的生产级治理功能•CWIC Flow 与云服务和数据源有更深入的集成
结论
LangGraph 为构建复杂多代理系统提供了强大的框架，具有结构化工作流、状态管理和条件路由。虽然 Clearwater Analytics 在 LangGraph 可用之前开发了 CWIC Flow，但我们继续评估新兴技术如 LangGraph 以增强我们的能力。
无论通过像 CWIC Flow 这样的定制解决方案还是像 LangGraph 这样的框架实现，多代理方法代表了 AI 系统的未来。通过协调具有不同角色和职责的专门代理，我们可以创建更强大、健壮和适应性的 AI 系统，为用户提供更大的价值。
随着 AI 技术的演进，Clearwater 致力于采用创新技术，增强我们为客户提供可操作见解的能力，同时保持行业领先的安全性、合规性和准确性标准。
