AutoGen
企业很少用
AgentScope
企业很少用
CAMEL
企业很少用
LangGraph
LangGraph 作为 LangChain 生态系统的重要扩展,代表了智能体框架设计的一个全新方向。与前面介绍的基于“对话”的框架(如 AutoGen 和 CAMEL)不同,LangGraph 将智能体的执行流程建模为一种状态机(State Machine),并将其表示为有向图(Directed Graph)。在这种范式中,图的节点(Nodes)代表一个具体的计算步骤(如调用 LLM、执行工具),而边(Edges)则定义了从一个节点到另一个节点的跳转逻辑。这种设计的革命性之处在于它天然支持循环,使得构建能够进行迭代、反思和自我修正的复杂智能体工作流变得前所未有的直观和简单。
要理解 LangGraph,我们需要先掌握它的三个基本构成要素。
首先,是全局状态(State)。整个图的执行过程都围绕一个共享的状态对象进行。这个状态通常被定义为一个 Python 的TypedDict,它可以包含任何你需要追踪的信息,如对话历史、中间结果、迭代次数等。所有的节点都能读取和更新这个中心状态。
fromtypingimportTypedDict,List# 定义全局状态的数据结构classAgentState(TypedDict):messages:List[str]# 对话历史current_task:str# 当前任务final_answer:str# 最终答案# ... 任何其他需要追踪的状态其次,是节点(Nodes)。每个节点都是一个接收当前状态作为输入、并返回一个更新后的状态作为输出的 Python 函数。节点是执行具体工作的单元。
# 定义一个“规划者”节点函数defplanner_node(state:AgentState)->AgentState:"""根据当前任务制定计划,并更新状态。"""current_task=state["current_task"]# ... 调用LLM生成计划 ...plan=f"为任务 '{current_task}' 生成的计划..."# 将新消息追加到状态中state["messages"].append(plan)returnstate# 定义一个“执行者”节点函数defexecutor_node(state:AgentState)->AgentState:"""执行最新计划,并更新状态。"""latest_plan=state["messages"][-1]# ... 执行计划并获得结果 ...result=f"执行计划 '{latest_plan}' 的结果..."state["messages"].append(result)returnstate最后,是边(Edges)。边负责连接节点,定义工作流的方向。最简单的边是常规边,它指定了一个节点的输出总是流向另一个固定的节点。而 LangGraph 最强大的功能在于条件边(Conditional Edges)。它通过一个函数来判断当前的状态,然后动态地决定下一步应该跳转到哪个节点。这正是实现循环和复杂逻辑分支的关键。
defshould_continue(state:AgentState)->str:"""条件函数:根据状态决定下一步路由。"""# 假设如果消息少于3条,则需要继续规划iflen(state["messages"])<3:# 返回的字符串需要与添加条件边时定义的键匹配return"continue_to_planner"else:state["final_answer"]=state["messages"][-1]return"end_workflow"在定义了状态、节点和边之后,我们可以像搭积木一样将它们组装成一个可执行的工作流。
fromlanggraph.graphimportStateGraph,END# 初始化一个状态图,并绑定我们定义的状态结构workflow=StateGraph(AgentState)# 将节点函数添加到图中workflow.add_node("planner",planner_node)workflow.add_node("executor",executor_node)# 设置图的入口点workflow.set_entry_point("planner")# 添加常规边,连接 planner 和 executorworkflow.add_edge("planner","executor")# 添加条件边,实现动态路由workflow.add_conditional_edges(# 起始节点"executor",# 判断函数should_continue,# 路由映射:将判断函数的返回值映射到目标节点{"continue_to_planner":"planner",# 如果返回"continue_to_planner",则跳回planner节点"end_workflow":END# 如果返回"end_workflow",则结束流程})# 编译图,生成可执行的应用app=workflow.compile()# 运行图inputs={"current_task":"分析最近的AI行业新闻","messages":[]}foreventinapp.stream(inputs):print(event)6.5.2 三步问答助手
在理解了 LangGraph 的核心概念之后,我们将通过一个实战案例来巩固所学。我们将构建一个简化的问答对话助手,它会遵循一个清晰、固定的三步流程来回答用户的问题:
- 理解 (Understand):首先,分析用户的查询意图。
- 搜索 (Search):然后,模拟搜索与意图相关的信息。
- 回答 (Answer):最后,基于意图和搜索到的信息,生成最终答案。
这个案例将清晰地展示如何定义状态、创建节点以及将它们线性地连接成一个完整的工作流。我们将代码分解为四个核心步骤:定义状态、创建节点、构建图、以及运行应用。
(1)定义全局状态
首先,我们需要定义一个贯穿整个工作流的全局状态。这是一个共享的数据结构,它在图的每个节点之间传递,作为工作流的持久化上下文。每个节点都可以读取该结构中的数据,并对其进行更新。
fromtypingimportTypedDict,Annotatedfromlanggraph.graph.messageimportadd_messagesclassSearchState(TypedDict):messages:Annotated[list,add_messages]user_query:str# 经过LLM理解后的用户需求总结search_query:str# 优化后用于Tavily API的搜索查询search_results:str# Tavily搜索返回的结果final_answer:str# 最终生成的答案step:str# 标记当前步骤我们创建了SearchState这个TypedDict,为状态对象定义了一个清晰的数据模式(Schema)。一个关键的设计是同时包含了user_query和search_query字段。这允许智能体先将用户的自然语言提问,优化成更适合搜索引擎的精炼关键词,从而显著提升搜索结果的质量。
(2)定义工作流节点
定义好状态结构后,下一步是创建构成我们工作流的各个节点。在 LangGraph 中,每个节点都是一个执行具体任务的 Python 函数。这些函数接收当前的状态对象作为输入,并返回一个包含更新后字段的字典。
在开始定义节点之前,我们先完成项目的初始化设置,包括加载环境变量和实例化大语言模型。
importosfromdotenvimportload_dotenvfromlangchain_openaiimportChatOpenAIfromlangchain_core.messagesimportHumanMessage,AIMessage,SystemMessagefromtavilyimportTavilyClient# 加载 .env 文件中的环境变量load_dotenv()# 初始化模型# 我们将使用这个 llm 实例来驱动所有节点的智能llm=ChatOpenAI(model=os.getenv("LLM_MODEL_ID","gpt-4o-mini"),api_key=os.getenv("LLM_API_KEY"),base_url=os.getenv("LLM_BASE_URL","https://api.openai.com/v1"),temperature=0.7)# 初始化Tavily客户端tavily_client=TavilyClient(api_key=os.getenv("TAVILY_API_KEY"))现在,我们来逐一创建三个核心节点。
(1) 理解与查询节点
此节点是工作流的第一步,此节点的职责是理解用户意图,并为其生成一个最优化的搜索查询。
defunderstand_query_node(state:SearchState)->dict:"""步骤1:理解用户查询并生成搜索关键词"""user_message=state["messages"][-1].content understand_prompt=f"""分析用户的查询:"{user_message}" 请完成两个任务: 1. 简洁总结用户想要了解什么 2. 生成最适合搜索引擎的关键词(中英文均可,要精准) 格式: 理解:[用户需求总结] 搜索词:[最佳搜索关键词]"""response=llm.invoke([SystemMessage(content=understand_prompt)])response_text=response.content# 解析LLM的输出,提取搜索关键词search_query=user_message# 默认使用原始查询if"搜索词:"inresponse_text:search_query=response_text.split("搜索词:")[1].strip()return{"user_query":response_text,"search_query":search_query,"step":"understood","messages":[AIMessage(content=f"我将为您搜索:{search_query}")]}该节点通过一个结构化的提示,要求 LLM 同时完成“意图理解”和“关键词生成”两个任务,并将解析出的专用搜索关键词更新到状态的search_query字段中,为下一步的精确搜索做好准备。
(2)搜索节点
该节点负责执行智能体的“工具使用”能力,它将调用 Tavily API 进行真实的互联网搜索,并具备基础的错误处理功能。
deftavily_search_node(state:SearchState)->dict:"""步骤2:使用Tavily API进行真实搜索"""search_query=state["search_query"]try:print(f"🔍 正在搜索:{search_query}")response=tavily_client.search(query=search_query,search_depth="basic",max_results=5,include_answer=True)# ... (处理和格式化搜索结果) ...search_results=...# 格式化后的结果字符串return{"search_results":search_results,"step":"searched","messages":[AIMessage(content="✅ 搜索完成!正在整理答案...")]}exceptExceptionase:# ... (处理错误) ...return{"search_results":f"搜索失败:{e}","step":"search_failed","messages":[AIMessage(content="❌ 搜索遇到问题...")]}此节点通过tavily_client.search发起真实的 API 调用。它被包裹在try...except块中,用于捕获可能的异常。如果搜索失败,它会更新step状态为"search_failed",这个状态将被下一个节点用来触发备用方案。
(3)回答节点
最后的回答节点能够根据上一步的搜索是否成功,来选择不同的回答策略,具备了一定的弹性。
defgenerate_answer_node(state:SearchState)->dict:"""步骤3:基于搜索结果生成最终答案"""ifstate["step"]=="search_failed":# 如果搜索失败,执行回退策略,基于LLM自身知识回答fallback_prompt=f"搜索API暂时不可用,请基于您的知识回答用户的问题:\n用户问题:{state['user_query']}"response=llm.invoke([SystemMessage(content=fallback_prompt)])else:# 搜索成功,基于搜索结果生成答案answer_prompt=f"""基于以下搜索结果为用户提供完整、准确的答案: 用户问题:{state['user_query']}搜索结果:\n{state['search_results']}请综合搜索结果,提供准确、有用的回答..."""response=llm.invoke([SystemMessage(content=answer_prompt)])return{"final_answer":response.content,"step":"completed","messages":[AIMessage(content=response.content)]}该节点通过检查state["step"]的值来执行条件逻辑。如果搜索失败,它会利用 LLM 的内部知识回答并告知用户情况。如果搜索成功,它则会使用包含实时搜索结果的提示,来生成一个有时效性且有据可依的回答。
(4)构建图
我们将所有节点连接起来。
fromlanggraph.graphimportStateGraph,START,ENDfromlanggraph.checkpoint.memoryimportInMemorySaverdefcreate_search_assistant():workflow=StateGraph(SearchState)# 添加节点workflow.add_node("understand",understand_query_node)workflow.add_node("search",tavily_search_node)workflow.add_node("answer",generate_answer_node)# 设置线性流程workflow.add_edge(START,"understand")workflow.add_edge("understand","search")workflow.add_edge("search","answer")workflow.add_edge("answer",END)# 编译图memory=InMemorySaver()app=workflow.compile(checkpointer=memory)returnapp(5)运行案例展示
运行此脚本后,您可以提出一些需要实时信息的问题,例如我们第一章中的案例:明天我要去北京,天气怎么样?有合适的景点吗
您会看到终端清晰地展示出智能体的“思考”过程:
🔍 智能搜索助手启动! 我会使用Tavily API为您搜索最新、最准确的信息 支持各种问题:新闻、技术、知识问答等 (输入 'quit' 退出) 🤔 您想了解什么: 明天我要去北京,天气怎么样?有合适的景点吗 ============================================================ 🧠 理解阶段: 我理解您的需求:理解:用户想了解明天北京的天气情况以及合适的景点推荐。 搜索词:北京 明天 天气 景点推荐 Beijing weather tomorrow attractions 🔍 正在搜索: 北京 明天 天气 景点推荐 Beijing weather tomorrow attractions 🔍 搜索阶段: ✅ 搜索完成!找到了相关信息,正在为您整理答案... 💡 最终回答: 明天(2025年9月17日)北京的天气预报显示,预计将是多云,气温范围在17°C(62°F)到25°C(77°F)之间。这种温和的天气非常适合户外活动。 ### 合适的景点推荐: 1. **长城**:作为中国最著名的历史遗址之一,长城是必游之地。你可以选择八达岭或慕田峪这些较为受欢迎的段落进行游览。 2. **故宫**:故宫是明清两代的皇宫,拥有丰富的历史和文化,适合对中国历史感兴趣的游客。 3. **天安门广场**:这是中国的象征之一,广场上有许多重要的建筑和纪念碑,适合拍照留念。 4. **颐和园**:一个非常美丽的皇家园林,适合漫步和欣赏自然风光,尤其是湖泊和古建筑。 5. **798艺术区**:如果你对现代艺术感兴趣,798艺术区是一个集艺术、文化和创意于一体的地方,适合探索和拍摄。 ### 小贴士: - 由于明天天气良好,建议提前规划出行路线,并准备一些水和小吃,以便在游览时保持充足的体力。 - 由于天气变化可能会影响游览体验,建议查看实时天气更新。 希望这些信息能帮助你安排一个愉快的北京之旅!如果你需要更多关于景点的信息或者旅行建议,欢迎随时询问。 ============================================================ 🤔 您想了解什么:并且他是一个可以持续交互的助手,你也可以继续向他发问。
6.5.3 LangGraph 的优势与局限性分析
任何技术框架都有其特定的适用场景和设计权衡。在本节中,我们将客观地分析 LangGraph 的核心优势及其在实际应用中可能面临的局限性。
(1)优势
如我们的智能搜索助手案例所示,LangGraph 将一个完整的实时问答流程,显式地定义为一个由状态、节点和边构成的“流程图”。这种设计的最大优势是高度的可控性与可预测性。开发者可以精确地规划智能体的每一步行为,这对于构建需要高可靠性和可审计性的生产级应用至关重要。其最强大的特性在于对循环(Cycles)的原生支持。通过条件边,我们可以轻松构建“反思-修正”循环,例如在我们的案例中,如果搜索失败,可以设计一个回退到备用方案的路径。这是构建能够自我优化和具备容错能力的智能体的关键。
此外,由于每个节点都是一个独立的 Python 函数,这带来了高度的模块化。同时,在流程中插入一个等待人类审核的节点也变得非常直接,为实现可靠的“人机协作”(Human-in-the-loop)提供了坚实的基础。
(2)局限性
与基于对话的框架相比,LangGraph 需要开发者编写更多的前期代码(Boilerplate)。定义状态、节点、边等一系列操作,使得对于简单任务而言,开发过程显得更为繁琐。开发者需要更多地思考“如何控制流程(how)”,而不仅仅是“做什么(what)”。由于工作流是预先定义的,LangGraph 的行为虽然可控,但也缺少了对话式智能体那种动态的、“涌现”式的交互。它的强项在于执行一个确定的、可靠的流程,而非模拟开放式的、不可预测的社会性协作。
调试过程同样存在挑战。虽然流程比对话历史更清晰,但问题可能出在多个环节:某个节点内部的逻辑错误、在节点间传递的状态数据发生异变,或是边跳转的条件判断失误。这要求开发者对整个图的运行机制有全局性的理解。