德阳市网站建设_网站建设公司_门户网站_seo优化
2026/1/16 18:04:34 网站建设 项目流程

LangGraph的StateGraph是一个用于构建状态图的核心组件,用于创建有状态的多步骤工作流程。以下是基本用法:

StateGraph基本概念

StateGraph允许你定义一个图结构,其中每个节点代表一个操作,边定义了节点之间的连接关系,整个图维护一个共享的状态。

基本用法步骤

1. 定义状态结构

from typing import TypedDictfrom langgraph.graph import StateGraph, ENDclass GraphState(TypedDict): input: str output: str step_count: int

2. 创建StateGraph实例

# 创建StateGraph,指定状态类型workflow = StateGraph(GraphState)

3. 定义节点函数

def node_1(state: GraphState) -> GraphState: """第一个节点的处理逻辑""" return { "input": state["input"], "output": f"处理了: {state['input']}", "step_count": state.get("step_count", 0) + 1 }def node_2(state: GraphState) -> GraphState: """第二个节点的处理逻辑""" return { "input": state["input"], "output": state["output"] + " -> 进一步处理", "step_count": state["step_count"] + 1 }

4. 添加节点

workflow.add_node("step1", node_1)workflow.add_node("step2", node_2)

5. 定义边连接

# 设置入口点workflow.set_entry_point("step1")# 添加边workflow.add_edge("step1", "step2")workflow.add_edge("step2", END)

6. 条件边的使用

你还可以添加条件边来实现分支逻辑:

def should_continue(state: GraphState) -> str: """决定下一步走向的条件函数""" if state["step_count"] < 3: return "continue" else: return "end"# 添加条件边workflow.add_conditional_edges( "step1", should_continue, { "continue": "step2", "end": END })

7. 编译并运行

# 编译图app = workflow.compile()# 运行图initial_state = { "input": "Hello World", "output": "", "step_count": 0}result = app.invoke(initial_state)print(result) ```![](http://cdn.zhipoai.cn/01b57ef5.jpg) **State基本概念** ------------- > 在现代AI应用开发中,如何优雅地管理复杂工作流的状态流转是一个关键挑战。LangGraph的State机制为我们提供了一套强大而灵活的状态管理解决方案。本文将深入探讨LangGraph State的设计理念、核心特性和实际应用。 ### **什么是LangGraph State?** LangGraph State是LangGraph框架中用于定义和管理图状态的核心组件。它不仅定义了数据的结构模式(Schema),还通过reducer函数指定了状态更新的逻辑。State既是所有节点和边的输入模式,也是节点间数据流转的载体。 核心设计理念 LangGraph State的设计遵循以下几个核心原则: * 类型安全: 通过TypedDict或Pydantic模型确保数据结构的正确性 * 状态隔离: 支持多层次的状态模式,实现输入、输出和内部状态的分离 * 灵活性: 节点可以读写任何已声明的状态通道 * 可扩展性: 支持运行时动态添加新的状态通道 ### **核心关系** 1. **`state_schema`**- 图的**内部状态模式**,定义了图内部所有可能的状态字段 2. **`input_schema`**- 图的**输入模式**,是`state_schema`的子集,定义图接受什么输入 3. **`output_schema`**- 图的**输出模式**,是`state_schema`的子集,定义图返回什么输出 ### **具体说明** ### **state\_schema(必需)** * 图的**完整内部状态**,包含所有节点可能读写的字段 * 所有节点都可以访问和写入这个schema中的任何字段 * 是图的"全局状态空间" ### **input\_schema(可选)** * 如果不指定,默认等于`state_schema` * 限制图的**输入接口**,只能传入这些字段 * 是`state_schema`的子集或相等 ### **output\_schema(可选)** * 如果不指定,默认等于`state_schema` * 限制图的**输出接口**,只返回这些字段 * 是`state_schema`的子集或相等 ### **实际例子** ```plaintext # 内部完整状态class OverallState(TypedDict): foo: str # 内部处理字段 user_input: str # 输入字段 graph_output: str # 输出字段# 限制输入只包含user_inputclass InputState(TypedDict): user_input: str# 限制输出只包含graph_output class OutputState(TypedDict): graph_output: str# 创建图builder = StateGraph( OverallState, # 内部完整状态 input_schema=InputState, # 输入限制 output_schema=OutputState # 输出限制)

关系图示

输入 → [InputState] → [OverallState] → [OutputState] → 输出 (输入过滤) (内部完整状态) (输出过滤)

这种设计的好处:

  • 封装性: 内部处理细节不会暴露给外部
  • 接口清晰: 明确定义输入输出契约
  • 灵活性: 内部可以有更多临时状态字段用于节点间通信

下面是一些例子

基础单模式使用

最简单的情况下,所有节点共享同一个状态模式:

from typing import TypedDictfrom langgraph.graph import StateGraph, START, ENDclass SimpleState(TypedDict): input: str output: str step_count: intdef process_node(state: SimpleState) -> SimpleState: return { "input": state["input"], "output": f"处理结果: {state['input']}", "step_count": state.get("step_count", 0) + 1 }# 创建简单的状态图workflow = StateGraph(SimpleState)workflow.add_node("processor", process_node)workflow.add_edge(START, "processor")workflow.add_edge("processor", END)app = workflow.compile()result = app.invoke({"input": "Hello", "output": "", "step_count": 0})

多模式架构:输入输出分离

在实际应用中,我们常常需要区分输入、输出和内部状态:

from typing import TypedDictfrom langgraph.graph import StateGraph, START, ENDclass InputState(TypedDict): """用户输入接口""" user_input: strclass OutputState(TypedDict): """最终输出接口""" graph_output: str step_count:intclass OverallState(TypedDict): """内部完整状态""" user_input: str intermediate_result: str processed_data: str graph_output: str step_count: intclass PrivateState(TypedDict): """节点间私有通信""" private_data: str internal_flag: booldef input_node(state: InputState) -> OverallState: """输入处理节点""" return { "user_input": state["user_input"], "intermediate_result": f"开始处理: {state['user_input']}", "step_count": 1 }def processing_node(state: OverallState) -> PrivateState: """数据处理节点""" processed = state["intermediate_result"].upper() return { "private_data": f"私有处理: {processed}", "internal_flag": len(state["user_input"]) > 5 }def intermediate_node(state: PrivateState) -> OverallState: """中间处理节点""" additional_info = " (长文本)" if state["internal_flag"] else " (短文本)" return { "processed_data": state["private_data"] + additional_info }def output_node(state: OverallState) -> OutputState: """输出生成节点""" final_result = f"最终结果: {state['processed_data']} | 步骤: {state.get('step_count', 0)}" return { "graph_output": final_result }# 创建多模式状态图builder = StateGraph( OverallState, input=InputState, output=OutputState)builder.add_node("input_processor", input_node)builder.add_node("data_processor", processing_node)builder.add_node("intermediate_processor", intermediate_node)builder.add_node("output_processor", output_node)builder.add_edge(START, "input_processor")builder.add_edge("input_processor", "data_processor")builder.add_edge("data_processor", "intermediate_processor")builder.add_edge("intermediate_processor", "output_processor")builder.add_edge("output_processor", END)graph = builder.compile()# 测试输入input_data = {"user_input": "Hello LangGraph","extra_input":"额外输入"}print("输入:", input_data) # 执行图result = graph.invoke(input_data)print("输出:", result)png_data=graph.get_graph().draw_mermaid_png()with open("state.png", "wb") as f: f.write(png_data)

输出如下

学AI大模型的正确顺序,千万不要搞错了

🤔2026年AI风口已来!各行各业的AI渗透肉眼可见,超多公司要么转型做AI相关产品,要么高薪挖AI技术人才,机遇直接摆在眼前!

有往AI方向发展,或者本身有后端编程基础的朋友,直接冲AI大模型应用开发转岗超合适!

就算暂时不打算转岗,了解大模型、RAG、Prompt、Agent这些热门概念,能上手做简单项目,也绝对是求职加分王🔋

📝给大家整理了超全最新的AI大模型应用开发学习清单和资料,手把手帮你快速入门!👇👇

学习路线:

✅大模型基础认知—大模型核心原理、发展历程、主流模型(GPT、文心一言等)特点解析
✅核心技术模块—RAG检索增强生成、Prompt工程实战、Agent智能体开发逻辑
✅开发基础能力—Python进阶、API接口调用、大模型开发框架(LangChain等)实操
✅应用场景开发—智能问答系统、企业知识库、AIGC内容生成工具、行业定制化大模型应用
✅项目落地流程—需求拆解、技术选型、模型调优、测试上线、运维迭代
✅面试求职冲刺—岗位JD解析、简历AI项目包装、高频面试题汇总、模拟面经

以上6大模块,看似清晰好上手,实则每个部分都有扎实的核心内容需要吃透!

我把大模型的学习全流程已经整理📚好了!抓住AI时代风口,轻松解锁职业新可能,希望大家都能把握机遇,实现薪资/职业跃迁~

这份完整版的大模型 AI 学习资料已经上传CSDN,朋友们如果需要可以微信扫描下方CSDN官方认证二维码免费领取【保证100%免费

需要专业的网站建设服务?

联系我们获取免费的网站建设咨询和方案报价,让我们帮助您实现业务目标

立即咨询