第三版：3、LangGraph之WorkFlow

WorkFlow介绍

工作流是 LLM 和工具通过预定义的代码路径进行编排的系统。另一方面，智能体是 LLM 动态指导其自身流程和工具使用的系统，它们都是构建复杂的大模型应用系统的核心组件。

LangGraph的工作流通过有向图(Directed Graph)定义，由节点（Node）、边（Edge）、状态（State）构成，支持条件分支、循环、并行执行。通过状态图（StartGraph）将多个智能体（Agent）和任务节点（Node）组织成可动态调整的流程结构，实现有状态、可扩展的任务编排。

LangGraph通过状态图将Agent和工作流解耦，让开发者能以“搭积木”的方式构建复杂AI系统:

• Agent是“乐高积木”，负责实现具体功能;
• 工作流是“搭建规则”，定义积木如何组合、何时执行。
  这种设计既保留了灵活性，又能通过状态管理实现大规模系统的可靠运行。

一句话总结：Agent 是一种特殊的节点(Node)，而工作流是整个图的运行流程(Graph)。Agent 负责"思考+行动”，工作流负责”编排+调度”。

相关概念

StateGraph

LangGraph 的核心是将 Agent 和 WorkFlow 建模为图。可以使用三个关键组件来定义一个Graph:

1. State：表示应用程序当前快照的共享数据结构。它可以是任何 Python 类型，但通常是TypedDict或Pydantic BaseModel。
2. Nodes：编码代理逻辑的 Python 函数。它们接收当前的State作为输入，执行一些计算或副作用，并返回更新后的State。
3. Edges：根据当前State决定接下来执行哪个Node的 Python 函数。它们可以是条件分支或固定转换。

构建流程

先定义状态，再添加节点和边，最后编译成图。

关键特性

节点和边可包含 LLM 或纯 Python 代码，支持构建复杂、循环的工作流，状态随时间演进。

State（状态）

表示应用当前快照的共享数据结构，通常是TypedDict或Pydantic模型。包含 schema 和 reducer 函数（决定如何应用更新）。

Reducer 函数

指定状态更新的方式（默认覆盖，也可自定义合并逻辑，如add_messages用于消息列表的智能更新）。

MessagesState

预构建的常用状态，包含消息列表，支持子类化扩展字段。

Node（节点）

执行具体逻辑的 Python 函数，接收状态（和可选配置）作为输入，返回状态更新。

特殊节点

• START：用户输入的入口节点。
• END：终止节点。

Edge（边）

决定节点间的路由逻辑。

• 普通边：固定从一个节点到另一个节点。
• 条件边：根据路由函数的返回值动态决定下一个节点（可并行执行多个节点）。路由函数接收当前状态，返回节点名称或映射值。

案例：评估器

使用LangGraph（LangChain 的一个扩展库）构建了一个循环式工作流（workflow），用于自动生成并评估笑话，如果笑话不够好笑，就基于反馈重新生成，直到达到“有趣”的标准为止。

一个典型的LLM Agent + 反馈循环（Feedback Loop）架构。

大致流程如下：

1. 用户提供一个主题（topic）
2. 系统生成一个笑话
3. LLM 对笑话进行评估（是否有趣 + 改进建议）
4. 如果不有趣 → 带着反馈重新生成
5. 如果有趣 → 结束流程
   

代码实现

创建项目

创建项目参考：第三版：1、LangGraph之基本介绍+项目生成

定义大模型

importosimportdotenvfromlangchain_openaiimportChatOpenAI dotenv.load_dotenv()llm=ChatOpenAI(api_key=os.getenv("OPENAI_API_KEY"),base_url=os.getenv("OPENAI_BASE_URL"),model="Qwen/Qwen2.5-72B-Instruct",)

核心代码实现

graph.py

fromtypingimportTypedDict,Literalfromlangchain_core.output_parsersimportStrOutputParserfromlanggraph.constantsimportSTART,ENDfromlanggraph.graphimportStateGraphfrompydanticimportBaseModel,Fieldfromagent.ai_modelimportllmclassState(TypedDict):joke:str# 笑话内容topic:str# 笑话主题feedback:str# 评估反馈funny_or_not:str# 评定是否为笑话# 结构化输出模型（用于LLM评估反馈）classFeedback(BaseModel):"""使用此工具来结构化响应内容"""grade:Literal["funny","not_funny"]=Field(examples=["funny","not_funny"],description="笑话是否有趣")feedback:str=Field(description="若不幽默，提供改进建议",examples=["可以加入双关语或意外结局"])# 节点函数defgenerator_joke(state:State):"""生成一个笑话"""prompt=(f"根据反馈改进笑话：{state['feedback']}\n，主题：{state['topic']}"ifstate.get("feedback",None)elsef"根据主题生成笑话：{state['topic']}")chain=llm|StrOutputParser()return{"joke":chain.invoke(prompt)}defevaluate_joke(state:State):"""评估笑话是否 有趣"""# 第一种写法（只适用于gpt等少量模型）：# chain = llm.with_structured_output(Feedback)# resp: Feedback = chain.invoke(# f"评估此笑话的幽默程度：\n{state['joke']}\n"# "注意：幽默应包含意外性或巧妙措辞"# )# return {"feedback": resp.feedback, "funny_or_not": resp.grade}# 第二种写法（通用）：chain=llm.bind_tools([Feedback])evaluation=chain.invoke(f"评估此笑话的幽默程度：\n{state['joke']}\n""注意：幽默应包含意外性或巧妙措辞")evaluation=evaluation.tool_calls[-1]["args"]print(f"evaluation:{evaluation}")return{"feedback":evaluation.get("feedback",""),"funny_or_not":evaluation.get("grade","not funny")}# 条件边的路由函数defroute_func(state:State):"""动态路由决策函数"""return("Accept"ifstate.get("funny_or_not",None)=="funny"else"Reject + Feedback")# 构建工作流builder=StateGraph(State)# 添加节点builder.add_node("joke_generator",generator_joke)builder.add_node("joke_evaluator",evaluate_joke)# 添加边builder.add_edge(START,"joke_generator")builder.add_edge("joke_generator","joke_evaluator")builder.add_conditional_edges("joke_evaluator",route_func,{"Accept":END,"Reject + Feedback":"joke_generator"})# 编译工作流graph=builder.compile()

核心代码解析

1. 类型定义：State

classState(TypedDict):joke:str# 当前生成的笑话topic:str# 用户提供的主题feedback:str# 上一轮评估给出的改进建议funny_or_not:str# "funny" 或 "not_funny"

• State是整个工作流的状态容器，所有节点共享这个状态。
• 每次节点执行后，会返回一个字典（如{"joke": "..."}），更新State中的部分字段。

2. 结构化输出模型：Feedback

classFeedback(BaseModel):grade:Literal["funny","not_funny"]=Field(...)feedback:str=Field(...)

• 这是一个Pydantic 模型，用于强制 LLM 输出结构化的评估结果。
• grade只能是"funny"或"not_funny"（通过Literal限制）。
• feedback字段用于提供改进建议（如“加入双关语”）。

💡 为什么需要结构化？
因为 LLM 默认输出是自由文本，但我们需要程序能可靠地提取“是否有趣”和“建议”，所以用工具调用（tool calling）或结构化输出来约束格式。

3. 节点函数

(1)generator_joke(state: State)

defgenerator_joke(state:State):prompt=(f"根据反馈改进笑话：{state['feedback']}\n，主题：{state['topic']}"ifstate.get("feedback",None)elsef"根据主题生成笑话：{state['topic']}")chain=llm|StrOutputParser()return{"joke":chain.invoke(prompt)}

• 返回{"joke": "..."}，更新状态中的joke字段。

(2)evaluate_joke(state: State)

defevaluate_joke(state:State):chain=llm.bind_tools([Feedback])evaluation=chain.invoke(f"评估此笑话的幽默程度：\n{state['joke']}\n""注意：幽默应包含意外性或巧妙措辞")evaluation=evaluation.tool_calls[-1]["args"]return{"feedback":evaluation.get("feedback",""),"funny_or_not":evaluation.get("grade","not funny")}

• 更新状态中的feedback和funny_or_not。

✅ 为什么用bind_tools而不是with_structured_output？

• with_structured_output仅支持部分模型（如 GPT-4o、Claude 3.5+ 等原生支持 JSON schema 的）。
• bind_tools是更通用的方式，几乎所有支持 function/tool calling 的模型都能用（包括开源模型如 Llama 3.1 + function calling 微调版）。

4. 路由函数：route_func

defroute_func(state:State):return"Accept"ifstate.get("funny_or_not")=="funny"else"Reject + Feedback"

• 根据评估结果决定下一步：
  • 如果funny_or_not == "funny"→ 走向END（接受）
  • 否则 → 回到joke_generator（拒绝并反馈）

注意：返回的字符串必须与add_conditional_edges中的 key 一致。

5. 构建工作流图（StateGraph）

builder=StateGraph(State)# 添加节点builder.add_node("joke_generator",generator_joke)builder.add_node("joke_evaluator",evaluate_joke)# 边连接builder.add_edge(START,"joke_generator")builder.add_edge("joke_generator","joke_evaluator")# 条件边（核心！实现循环）builder.add_conditional_edges("joke_evaluator",route_func,{"Accept":END,"Reject + Feedback":"joke_generator"})graph=builder.compile()

6. 图结构

START ↓ joke_generator → joke_evaluator ↘ (if not funny) ↖___________ (loop back) ↘ (if funny) → END

• add_conditional_edges是实现动态路由/循环的关键。
• 整个工作流可以执行多次迭代，直到笑话被判定为“funny”。

7. 潜在改进点

1. 防止无限循环：

   • 可添加最大重试次数（如最多3次），避免 LLM 一直生成不好笑的笑话。
   • 方法：在State中加attempt_count，在route_func中判断。

2. 初始状态简化：

   • 调用时只需传topic，其他字段可设默认值（可通过__init__或预处理实现）。

效果演示

代码仓地址

langgraph-workflow-evaluator

自此，本文分享到此结束！！！