LangGraph Agent架构实战：构建具备动态规划与执行能力的智能体工作流

1. LangGraph Agent架构概述

LangGraph是一个专注于构建有状态、多角色应用程序的库，它利用大语言模型（LLMs）来创建智能体和多智能体工作流。这个框架的核心优势体现在以下几个方面：

周期性支持：LangGraph允许开发者定义包含循环的流程，这对大多数智能体架构至关重要。这种能力使得LangGraph能够处理需要重复步骤或迭代循环的复杂任务，而不仅仅是基于有向无环图（DAG）的解决方案。

高度可控性：LangGraph提供了对应用程序流程和状态的精细控制。这种精细控制对于创建行为可预测、符合预期的智能体至关重要，特别是在处理复杂或敏感的应用场景时。

持久性功能：LangGraph内置了持久性功能，这意味着智能体能够跨交互保持上下文和记忆。这对于实现长期任务的一致性和连续性非常关键。持久性还支持高级的人机交互，允许人类输入无缝集成到工作流程中，使智能体能够通过记忆功能学习和适应。

2. LangGraph核心特性解析

2.1 循环与分支

循环和分支是LangGraph最强大的特性之一。它允许在应用程序中实现循环和条件语句，适用于需要重复执行任务或根据不同条件执行不同操作的场景，如自动化决策流程、复杂业务逻辑处理等。

在实际应用中，这意味着你可以构建这样的智能体：

• 自动生成代码并自我修正，直到通过测试
• 执行Web导航时根据页面内容动态调整操作
• 处理多步骤任务时根据中间结果改变执行路径

2.2 状态管理

LangGraph的状态管理是其核心优势之一。状态是一个共享的数据结构，存储了所有与当前执行上下文相关的信息。状态可以是任何Python类型，通常使用TypedDict或PydanticBaseModel。

状态管理的关键特点：

• 自动维护：当一个节点执行并返回更新后的状态时，框架确保新状态被传递到下一个节点
• 生命周期：状态的生命周期与图的执行周期相匹配，从图的初始状态开始，在每个节点执行时更新，直到图执行结束
• 通信机制：状态提供了节点间的通信机制，每个节点可以读取前一个节点更新的状态，并在此基础上操作

2.3 节点与边

在LangGraph框架中，节点（Nodes）是Python函数，它们编码了智能体的逻辑。节点的第一个位置参数是State，第二个可选参数是config（节点对应的处理逻辑）。

边（Edges）是连接节点的对象，它们定义了节点之间的转换逻辑。每条边都连接两个节点：一个源节点和一个目标节点。边的主要功能是确定何时应该从源节点跳转到目标节点。

LangGraph支持两种类型的边：

• 正常边：直接从节点A跳转到节点B
• 条件边：调用一个函数来确定下一步应该跳转到哪个节点

3. 构建Plan-and-Execute智能体

3.1 设计思路

Plan-and-Execute的核心思想是：首先针对用户的目标提出一系列具体操作，之后依次完成各个操作，如果未达成目标则会重新规划。这种架构的优势在于：

1. 明确的长期规划，有利于在执行规划中使用最好的模型
2. 在执行各个任务过程中，可以使用较弱的模型以节省成本
3. 动态调整能力，根据执行结果优化后续步骤

3.2 实现步骤

典型的Plan-and-Execute工作流包含以下步骤：

1. 用户提出需求
2. 针对需求进行规划
3. 生成一系列任务列表
4. 执行每一项任务（这里使用到各个任务对应的智能体）
5. 任务执行结束，更新状态（状态用于确认用户需求是否完成）
6. 如果达成用户需求，则返回结果给用户，否则，基于当前状态重新生成任务列表，跳转到第3步

3.3 关键组件实现

3.3.1 状态定义

我们首先定义一个PlanExecute类来描述执行计划的结构：

from typing import Annotated, List, Tuple from typing_extensions import TypedDict import operator class PlanExecute(TypedDict): input: str # 原始输入，字符串类型 plan: List[str] # 构建计划列表 past_steps: Annotated[List[Tuple], operator.add] # 追踪之前的执行步骤，数据类型为元组列表 response: str # 最终输出，字符串类型

3.3.2 规划节点

规划节点负责根据用户输入生成初始计划：

from pydantic import BaseModel, Field class Plan(BaseModel): """要遵循的未来计划""" steps: List[str] = Field( description="要遵循的不同步骤，应按排序顺序排列" ) planner_prompt = ChatPromptTemplate.from_messages([ ("system", '''针对给定目标，提出一个简单的分步计划。 这个计划应包含单独的任务，如果正确执行将产生正确答案。 不要添加任何多余的步骤。 最后一步的结果应该是最终答案。 确保每个步骤都包含所需的所有信息 - 不要跳过步骤。'''), ("placeholder", "{messages}"), ]) planner = planner_prompt | llm.with_structured_output(Plan)

3.3.3 执行节点

执行节点负责实际执行计划中的任务：

async def execute_step(state: PlanExecute): plan = state["plan"] plan_str = "\n".join(f"{i+1}. {step}" for i, step in enumerate(plan)) task = plan[0] task_formatted = f"""对于以下计划： {plan_str} 你的任务是执行第1步：{task}。""" agent_response = await agent_executor.ainvoke( {"messages": [("user", task_formatted)]} ) return { "past_steps": [(task, agent_response["messages"][-1].content)], }

3.3.4 重新规划节点

重新规划节点根据当前状态决定是继续执行还是结束：

from typing import Union class Response(BaseModel): """给用户的响应""" response: str class Act(BaseModel): """要执行的动作""" action: Union[Response, Plan] = Field( description="要执行的动作。如果想响应用户，使用Response。" "如果需要进一步使用工具获取答案，使用Plan。" ) replanner_prompt = ChatPromptTemplate.from_template( """针对给定目标，提出一个简单的分步计划。 这个计划应包含单独的任务，如果正确执行将产生正确答案。 不要添加任何多余的步骤。 最后一步的结果应该是最终答案。 确保每个步骤都包含所需的所有信息 - 不要跳过步骤。 你的原始目标是：{input} 你的原始计划是：{plan} 你目前已完成以下步骤：{past_steps} 相应地更新你的计划。如果不需要更多步骤并且可以返回给用户，那么就用那个响应。 否则，填写计划。只添加计划中仍然需要完成的步骤。不要将之前完成的步骤作为计划的一部分返回。""" ) replanner = replanner_prompt | llm.with_structured_output(Act)

4. 构建完整工作流

4.1 图结构定义

将各个节点组合成完整的工作流：

from langgraph.graph import StateGraph, START workflow = StateGraph(PlanExecute) # 添加规划节点 workflow.add_node("planner", plan_step) # 添加执行节点 workflow.add_node("agent", execute_step) # 添加重新规划节点 workflow.add_node("replan", replan_step) # 从开始到规划节点 workflow.add_edge(START, "planner") # 从规划到执行节点 workflow.add_edge("planner", "agent") # 从执行到重新规划节点 workflow.add_edge("agent", "replan") # 条件边控制流程 workflow.add_conditional_edges( "replan", should_end, ["agent", END], ) app = workflow.compile()

4.2 执行流程

完整的执行流程如下：

1. 用户提出目标
2. 目标被拆分成多个步骤
3. 开始执行第一个步骤
4. 执行完成后，将智能体的回答加入到past_steps中
5. 使用replan更新计划
6. 如果已完成目标，则结束，否则更新计划后循环到第3步执行

4.3 实际应用示例

让我们看一个实际的市场调研报告生成示例：

inputs = { "input": "调研2024年新能源汽车市场的三大趋势，并给出简要分析" } async def main(inputs, config): async for event in app.astream(inputs, config=config): for k, v in event.items(): if k != "__end__": print(v) config = {"recursion_limit": 50} asyncio.run(main(inputs, config))

这个智能体会自动执行以下步骤：

1. 识别新能源汽车市场的关键指标
2. 搜索最新的市场报告和数据
3. 分析并总结出三大趋势
4. 验证信息的准确性
5. 生成结构化的调研报告

如果在任何步骤中发现信息不足或分析不充分，智能体会自动调整计划，补充新的调研任务，直到生成满意的报告为止。

5. 高级技巧与优化建议

5.1 性能优化

对于复杂的任务，可以考虑以下优化策略：

1. 并行执行：修改图结构，使不依赖的任务可以并行执行
2. 缓存机制：对重复的子任务结果进行缓存
3. 模型选择：对规划节点使用强大但昂贵的模型，对执行节点使用经济型模型

5.2 错误处理

健壮的智能体需要处理各种异常情况：

async def safe_execute_step(state: PlanExecute): try: return await execute_step(state) except Exception as e: return { "error": str(e), "past_steps": [("failed_step", f"Step failed with error: {str(e)}")] }

5.3 人机协作

集成人工审核节点，在关键决策点引入人工干预：

async def human_review_step(state: PlanExecute): important_data = state.get("critical_data") send_for_review(important_data) await wait_for_review() return { "human_feedback": get_review_result() }

5.4 评估与监控

建立评估机制来监控智能体性能：

def evaluate_performance(state: PlanExecute): steps_taken = len(state["past_steps"]) time_used = time.time() - state["start_time"] accuracy = calculate_accuracy(state["response"]) return { "performance": { "steps": steps_taken, "time": time_used, "accuracy": accuracy } }

6. 实际案例：市场调研智能体

让我们深入构建一个市场调研智能体的具体实现：

6.1 状态定义

class MarketResearchState(TypedDict): research_topic: str objectives: List[str] data_sources: List[str] collected_data: Dict[str, Any] analysis: str report: str validation: List[str] completed: bool

6.2 规划节点

research_planner_prompt = ChatPromptTemplate.from_messages([ ("system", '''你是一个市场调研专家。针对给定的调研主题，请制定一个详细的调研计划。 计划应包含： 1. 明确的调研目标（3-5个） 2. 推荐的数据来源 3. 分析方法 4. 报告结构建议'''), ("user", "{research_topic}") ]) class ResearchPlan(BaseModel): objectives: List[str] data_sources: List[str] analysis_methods: List[str] report_structure: List[str] research_planner = research_planner_prompt | llm.with_structured_output(ResearchPlan)

6.3 数据收集节点

async def collect_data(state: MarketResearchState): tools = [ TavilySearchResults(max_results=5), FinancialDataTool(), IndustryReportTool() ] agent = create_react_agent(llm, tools) collected = {} for source in state["data_sources"]: result = await agent.ainvoke({ "messages": [("user", f"收集关于{state['research_topic']}的{source}数据")] }) collected[source] = result["messages"][-1].content return {"collected_data": collected}

6.4 分析节点

async def analyze_data(state: MarketResearchState): analysis_prompt = ChatPromptTemplate.from_messages([ ("system", '''你是一个资深市场分析师。请根据以下数据进行分析： 调研主题：{research_topic} 调研目标：{objectives} 数据： {collected_data} 请提供详细的分析，重点关注趋势、模式和关键发现。'''), ]) analyzer = analysis_prompt | llm analysis = await analyzer.ainvoke(state) return {"analysis": analysis.content}

6.5 报告生成节点

async def generate_report(state: MarketResearchState): report_prompt = ChatPromptTemplate.from_messages([ ("system", '''根据以下分析生成专业市场调研报告： 调研主题：{research_topic} 分析结果： {analysis} 请按照以下结构组织报告： 1. 执行摘要 2. 主要发现 3. 市场趋势 4. 建议 5. 结论'''), ]) reporter = report_prompt | llm report = await reporter.ainvoke(state) return {"report": report.content, "completed": True}

6.6 构建完整工作流

research_workflow = StateGraph(MarketResearchState) research_workflow.add_node("plan", research_plan_step) research_workflow.add_node("collect", collect_data) research_workflow.add_node("analyze", analyze_data) research_workflow.add_node("report", generate_report) research_workflow.add_edge(START, "plan") research_workflow.add_edge("plan", "collect") research_workflow.add_edge("collect", "analyze") research_workflow.add_edge("analyze", "report") research_workflow.add_edge("report", END) research_app = research_workflow.compile()

这个市场调研智能体可以处理复杂的调研任务，自动完成从规划到报告生成的全过程，大大提高了市场调研的效率和准确性。