企业官网建设流程全解析

网站建设公司十年乐云seo,中核华兴建设有限公司网站,网站建设的摘要,建筑网方成龙LangGraph是解决LLM幻觉问题的关键工具#xff0c;通过支持循环逻辑、状态管理和人机协作等核心优势#xff0c;超越了传统LangChain的DAG限制。文章详细介绍了LangGraph的结构组成#xff08;全局状态、节点、边#xff09;#xff0c;并通过代码示例展示了智能翻译Agent…LangGraph是解决LLM幻觉问题的关键工具通过支持循环逻辑、状态管理和人机协作等核心优势超越了传统LangChain的DAG限制。文章详细介绍了LangGraph的结构组成全局状态、节点、边并通过代码示例展示了智能翻译Agent的构建。此外还深入探讨了状态管理、持久化机制以及如何实现ReAct、Plan-and-Solve等多种Agent执行范式为开发者提供了构建高可靠性AI应用的完整解决方案。像经常用 LLM 的同学都知道现在最头疼的问题就是幻觉问题在金融或精密计算领域不确定性意味着风险。 如果 Agent 负责分析 NVDA 或 TSLA 的财报开发者希望它在处理相同数据时逻辑推导链条是严密的而不是在不同时间给出自相矛盾的结论。或是需要 LLM 输出 JSON 来触发一个 API我们不会希望 LLM 在 JSON 里多加了一个逗号或改变了字段名。最后我还尝试用 LangGraph 的理念自己写了一个 smallest-LangGraph LangGraph 可以做什么传统的 LangChain 核心逻辑是DAG有向无环图。我们可以轻松定义A - B - C的步骤但如果你想让 AI 在B步骤发现结果不满意自动跳回A重新执行LangChain 的普通 Chain 很难优雅地实现。并且在复杂的长对话或多步骤任务中维护一个全局的、可持久化的“状态快照”非常困难。所以为了解决这些问题LangGraph 就诞生了。LangGraph 的主要有这些核心优势支持“循环Cycles”与“迭代”思考- 2.行动- 3.观察结果- 4.如果不满意回到第1步。 LangGraph 允许你定义这种闭环逻辑这在长任务、自我修正代码、多轮调研场景下是刚需。状态管理LangGraph 引入了State的概念所有节点共享同一个TypedDict你可以精确定义哪些数据是追加的operator.add哪些是覆盖的。并且它可以自动保存每一步的状态。即使程序崩溃或需要人工审核你也可以从特定的“存档点”恢复而不需要从头运行。人机协作LangGraph 允许你将流程设计为“在某处强制停下”等待人类信号后再继续。这在 LangChain 的线性模型中极难实现但在 LangGraph 的状态机模型中只是一个节点属性。高度可控“如果工具返回报错必须走 A 路径。” 这种确定性对于生产环境的后端服务至关重要。不能让模型乱输出在生产环境上严格把控输出结果是很重要的。LangGraph 结构由于 LangGraph 的核心思想是将 Agent 的工作流建模为一张有向图Directed Graph。所以 LangGraph 有如下几个结构组成全局状态State这个状态通常被定义为一个 Python 的TypedDict它可以包含任何你需要追踪的信息如对话历史、中间结果、迭代次数等所有的节点都能读取和更新这个中心状态。节点Nodes每个节点都是一个接收当前状态作为输入、并返回一个更新后的状态作为输出的 Python 函数。边Edges边负责连接节点定义工作流的方向。最简单的边是常规边它指定了一个节点的输出总是流向另一个固定的节点。而 LangGraph 最强大的功能在于条件边Conditional Edges。它通过一个函数来判断当前的状态然后动态地决定下一步应该跳转到哪个节点。基于上面的概念我们来做一个例子假设我们要开发一个 Agent它先翻译一段话然后自己检查是否有语法错误如果有就打回重新翻译如果没有就结束。首先我们先定义状态 (State)from typing import TypedDict, Listclass AgentState(TypedDict): # 原始文本 input_text: str # 翻译后的文本 translated_text: str # 反思反馈 feedback: str # 循环次数防止死循环 iterations: int定义节点逻辑 (Nodes)def translator_node(state: AgentState): print(--- 正在翻译 ---) # 这里通常会调用 LLM new_text fTranslated: {state[input_text]} return {translated_text: new_text, iterations: state.get(iterations, 0) 1}def critic_node(state: AgentState): print(--- 正在自检 ---) # 模拟检查逻辑如果包含 bad 字符就认为不合格 if bad in state[translated_text]: return {feedback: 发现不当词汇请重试} return {feedback: OK}定义路由逻辑 (Conditional Edges)def should_continue(state: AgentState): if state[feedback] OK or state[iterations] 3: return end else: return rephrase构建图 (Graph Construction)from langgraph.graph import StateGraph, END# 1. 初始化图workflow StateGraph(AgentState)# 2. 添加节点workflow.add_node(translator, translator_node)workflow.add_node(critic, critic_node)# 3. 设置入口点workflow.set_entry_point(translator)# 4. 连接节点workflow.add_edge(translator, critic)# 5. 添加条件边 (根据 critic 的反馈决定去向)workflow.add_conditional_edges( critic, should_continue, { rephrase: translator, # 如果不 OK回到翻译节点 end: END # 如果 OK结束 })# 6. 编译成可执行应用app workflow.compile()通过上面这种编排方式可以让 LLM 概率性输出产生确定性的输出通过各种限制节点很好的控制了 LLM 的访问的节点。下面我给出完整的例子大家可以用这个例子去尝试一下from typing import TypedDict, Listfrom langchain_openai import ChatOpenAIfrom langgraph.graph import StateGraph, ENDfrom langchain_core.messages import SystemMessage, HumanMessagellm ChatOpenAI( temperature0.6, modelglm-4.6V, openai_api_key, openai_api_basehttps://open.bigmodel.cn/api/paas/v4/)class AgentState(TypedDict): # 原始文本 input_text: str # 翻译后的文本 translated_text: str # 反思反馈 feedback: str # 循环次数防止死循环 iterations: intdef translator_node(state: AgentState): 翻译节点负责将中文翻译成英文 print(f\n--- [节点翻译器] 第 {state.get(iterations, 0) 1} 次尝试 ---) iters state.get(iterations, 0) feedback state.get(feedback, 无) # 构建提示词如果是重试带上反馈建议 system_prompt 你是一个专业的翻译官。请将用户的中文翻译成地道、优雅的英文。 if iters 0: system_prompt f 注意这是第二次尝试请参考之前的反馈进行改进{feedback} response llm.invoke([ SystemMessage(contentsystem_prompt), HumanMessage(contentstate[input_text]) ]) return { translated_text: response.content, iterations: iters 1 }def critic_node(state: AgentState): 评审节点检查翻译质量 print(--- [节点评审员] 正在检查翻译质量... ---) system_prompt ( 你是一个严苛的英文编辑。请评价以下翻译是否准确、地道。 如果翻译得很好请只回复关键词【PASS】。 如果翻译有改进空间请直接指出问题并给出改进建议。 ) user_content f原文{state[input_text]}\n译文{state[translated_text]} response llm.invoke([ SystemMessage(contentsystem_prompt), HumanMessage(contentuser_content) ]) return {feedback: response.content}# 4. 定义路由逻辑def should_continue(state: AgentState): 判断是继续修改还是直接结束 if【PASS】in state[feedback] or state[iterations] 3: if state[iterations] 3: print(!!! 达到最大尝试次数停止优化。) returnend else: print(f 反馈建议{state[feedback]}) returnrephrase# 1. 初始化图workflow StateGraph(AgentState)# 2. 添加节点workflow.add_node(translator, translator_node)workflow.add_node(critic, critic_node)# 3. 设置入口点workflow.set_entry_point(translator)# 4. 连接节点workflow.add_edge(translator, critic)# 5. 添加条件边 (根据 critic 的反馈决定去向)workflow.add_conditional_edges( critic, should_continue, { rephrase: translator, # 如果不 OK回到翻译节点 end: END # 如果 OK结束 })# 6. 编译成可执行应用app workflow.compile()# 7. 运行时交互if __name__ __main__: print( LangGraph 智能翻译 Agent (输入 exit 退出) ) whileTrue: user_input input(\n请输入想要翻译的中文内容: ) if user_input.lower() exit: break # 初始状态 initial_state { input_text: user_input, iterations: 0 } # 运行图并获取最终状态 final_state app.invoke(initial_state) print(\n * 30) print(f最终翻译结果\n{final_state[translated_text]}) print( * 30)LangGraph 是如何管理状态的State Reducer 自动合并 stateReducer 在 LangGraph 中就是一种更新状态的处理逻辑如果没有指定默认行为是用新值覆盖旧值。想要指定 Reducer 只需要通过 typing.Annotated 字段绑定一个 Reducer 函数即可。比如使用 operator.add 定义这是一个“追加型”字段from typing import Annotated, TypedDictfrom langgraph.graph import StateGraph, ENDimport operator# 定义状态结构 (类似 Go 的 Struct)class AgentState(TypedDict): # 使用 Annotated 和 operator.add 定义这是一个“追加型”字段 # 每次节点返回消息都会 append 到这个列表而不是覆盖它 messages: Annotated[list[str], operator.add] # 普通字段默认行为是 Overwrite (覆盖) # 适合存储状态机当前的步骤或分析结论 current_status: str # 计数器也可以使用 operator.add 实现增量累加 retry_count: Annotated[int, operator.add]Checkpointer Thread 持久化状态在 LangGraph 中Checkpointer是一个持久化层接口这意味着历史的对话记录可以被自动持久化到数据库如 SQLite 或其他外部数据库中。这使得即使应用程序重启或用户断开连接对话历史也能被保存和恢复从而实现“真正的多轮记忆”。LangGraph 提供了多种 Checkpointer 以便应对不同的使用场景MemorySaver 保存在内存适用开发调试、单元测试SqliteSaver 保存在本地的.db文件轻量级应用、边缘计算适合单机部署PostgresSaver 保存在 PostgreSQL适合用在生产环境、多实例部署RedisSaver 适合处理高频、短时会话LangGraph 通过 thread_id 会话的唯一标识结合 Checkpointer 就可以实现状态的隔离首先指定一个 指定一个thread_id所有相关的状态都会被保存到这个线程中。config {configurable: {thread_id: conversation_1}}graph.invoke(input_data, config)编译的时候传入 Checkpointer 即可。# 创建 checkpointercheckpointer InMemorySaver()# 编译图时传入 checkpointergraph builder.compile(checkpointercheckpointer)完整示例from langgraph.checkpoint.memory import InMemorySaverfrom langchain_openai import ChatOpenAI from langgraph.graph import StateGraph, START, END, MessagesState llm ChatOpenAI( temperature0.6, modelglm-4.6v, openai_api_key, openai_api_basehttps://open.bigmodel.cn/api/paas/v4/)# 定义节点函数def call_model(state: MessagesState): response llm.invoke(state[messages]) return {messages: response}# 构建图builder StateGraph(MessagesState)builder.add_node(agent, call_model)builder.add_edge(START, agent)# 创建 checkpointercheckpointer InMemorySaver()# 编译图时传入 checkpointergraph builder.compile(checkpointercheckpointer)# 第一次对话config {configurable: {thread_id: user_123}}response1 graph.invoke( {messages: [{role: user, content: 你好我的名字是张三}]}, config)print(fAI: {response1[messages][-1].content})# 第二次对话相同 thread_idresponse2 graph.invoke( {messages: [{role: user, content: 我的名字是什么}]}, config # 使用相同的 thread_id)print(fAI: {response2[messages][-1].content})# 获取当前的状态信息print(fAI: {graph.get_state(config)})除此之外可以graph.get_state()/graph.get_state_history()拿到当前/历史状态也可以基于 checkpoint 做 replay、update_state时间旅行能力通常要求启用 checkpointer。Super-step 原子循环单元由于一个 node 也可以连接多个 node多个 node 也可以连接到 一个 node所以 LangGraph 设计了 Super-step 来作为原子循环单元。比如下面的例子graph.set_entry_point(n1) graph.add_edge(n1, n2) graph.add_edge(n1, n3) graph.add_edge(n2, n4) graph.add_edge(n3, n4) graph.add_edge(n4, END)LangGraph 只分了三步就执行完了该循环。如下图第二步的时候会 n2、n3 节点并行执行。graph active nodes in each superstep并且每个 super-step 都会自动保存一个 checkpoint这就是持久化机制的基础。即使程序中断也能从最后一个 super-step 的 checkpoint 恢复执行。Human-in-the-loop 人机协同Human-in-the-loop 本质上就是让 agent “关键时刻”暂停它的底层靠的是 **interrupt 持久化(checkpoint)**暂停时把状态存起来恢复时从存档续跑。比如我们想要是线一个场景就是让 AI 去判断是否应该要人工审核如过需要人工审核那么就 interrupt 进行中断然后等人工输入之后根据执行逻辑进行恢复然后配合Command(resume...)恢复。基本流程可以是这样import uuidfrom langgraph.types import interrupt, Commanddef ask_human(state): answer interrupt(Do you approve?) return {approved: answer}config {configurable: {thread_id: str(uuid.uuid4())}}# 第一次跑会中断返回 __interrupt__graph.invoke({input: x}, configconfig)# 人给了答复后用 Command(resume...) 恢复graph.invoke(Command(resumeTrue), configconfig)这个例子中interrupt()会暂停图执行把一个值必须可 JSON 序列化抛给调用方并依赖 checkpointer 持久化状态然后你用同一个thread_id重新调用图并传入Command(resume...)来继续。接下来我们看一个完整的例子设计一个常见的场景当模型觉得需要“找专家/找人类”时会调用一个工具human_assistance而这个工具会用interrupt()把流程暂停下来等你在命令行里输入专家建议后再用Command(resume...)把图唤醒继续跑。from langgraph.checkpoint.memory import InMemorySaverfrom langchain_openai import ChatOpenAIfrom typing_extensions import TypedDictfrom typing import Annotatedfrom langgraph.graph import StateGraph, START, ENDfrom langgraph.graph.message import add_messagesfrom langgraph.types import Command, interruptfrom langchain_core.tools import toolfrom langgraph.prebuilt import ToolNode, tools_condition from langchain_core.messages import HumanMessagellm ChatOpenAI( temperature0.6, modelglm-4.6v, openai_api_key, openai_api_basehttps://open.bigmodel.cn/api/paas/v4/)class State(TypedDict): messages: Annotated[list, add_messages]tooldef human_assistance(query: str) - str: Request assistance from a human. human_response interrupt({query: query}) return human_response[data]tools [human_assistance]llm_with_tools llm.bind_tools(tools)def chatbot(state: State): message llm_with_tools.invoke(state[messages]) return {messages: [message]}tool_node ToolNode(toolstools)graph_builder StateGraph(State)graph_builder.add_node(tools, tool_node)graph_builder.add_conditional_edges( chatbot, tools_condition,)graph_builder.add_node(chatbot, chatbot)graph_builder.add_edge(START, chatbot)graph_builder.add_edge(chatbot, END)memory InMemorySaver()graph graph_builder.compile(checkpointermemory)config {configurable: {thread_id: test_thread_123}}# 第一步用户提出一个需要“人工协助”的问题print(--- 第一阶段AI 运行并遇到 interrupt ---)initial_input HumanMessage(content你好帮我找个专家回答我的问题)for event in graph.stream({messages: [initial_input]}, config, stream_modevalues): ifmessagesin event: event[messages][-1].pretty_print()# 此时你会发现程序停止了因为它卡在 human_assistance 的 interrupt 处。# 第二步模拟人类你在一段时间后看到了请求并回复print(\n--- 第二阶段模拟人类介入并提供答案 ---)# 我们构造一个 Command 对象来“唤醒”它# resume 里的内容会直接成为 interrupt() 函数的返回值expert_input input(专家建议: )human_feedback {data: expert_input}for event in graph.stream( Command(resumehuman_feedback), # 这里是恢复运行的关键 config, stream_modevalues): ifmessagesin event: event[messages][-1].pretty_print()snapshot graph.get_state(config)print(snapshot.values)LangGraph如何轻松实现 Agent 多种执行范式ReActReAct由Shunyu Yao于2022年提出[1]其核心思想是模仿人类解决问题的方式将推理 (Reasoning)与行动 (Acting)显式地结合起来形成一个“思考-行动-观察”的循环。imgReAct范式通过一种特殊的提示工程来引导模型使其每一步的输出都遵循一个固定的轨迹Thought (思考)这是智能体的“内心独白”。它会分析当前情况、分解任务、制定下一步计划或者反思上一步的结果。Action (行动)这是智能体决定采取的具体动作通常是调用一个外部工具API 。Observation (观察)这是执行Action后从外部工具返回的结果例如或API的返回值。智能体将不断重复这个Thought - Action - Observation的循环将新的观察结果追加到历史记录中形成一个不断增长的上下文直到它在Thought中认为已经找到了最终答案然后输出结果。from typing import Annotated, Literalfrom typing_extensions import TypedDictfrom langgraph.graph import StateGraph, ENDfrom langgraph.graph.message import add_messagesfrom langgraph.prebuilt import ToolNodefrom langchain_openai import ChatOpenAIfrom langchain_core.tools import tool# --- 1. 双手定义查天气的工具 ---tooldef get_weather(city: str): 查询指定城市的天气 # 这里模拟后端 API 返回数据 if北京in city: return晴天25度 return阴天20度tools [get_weather]tool_node ToolNode(tools)# --- 2. 记忆定义存储对话的状态 ---class State(TypedDict): messages: Annotated[list, add_messages]# --- 3. 大脑定义思考逻辑 ---model ChatOpenAI( temperature0.6, modelglm-4.6v, openai_api_key, openai_api_basehttps://open.bigmodel.cn/api/paas/v4/).bind_tools(tools)def call_model(state: State): # 大脑看一眼目前的对话决定是直接说话还是去用手拿工具 return {messages: [model.invoke(state[messages])]}# --- 4. 路由判断下一步是干活还是结束 ---def should_continue(state: State): last_message state[messages][-1] # 如果大脑发出的指令包含“调用工具”就去 tools 节点 if last_message.tool_calls: returntools # 如果大脑直接说话了就结束 return END# --- 5. 编排图把脑和手连起来 ---workflow StateGraph(State)workflow.add_node(agent, call_model)workflow.add_node(tools, tool_node)workflow.set_entry_point(agent)# 条件边agent 运行完判断是去 tools 还是结束workflow.add_conditional_edges(agent, should_continue)# 普通边tools 运行完干完活了必须把结果拿回给 agent 看workflow.add_edge(tools, agent)app workflow.compile()# --- 6. 执行测试 ---for chunk in app.stream({messages: [(user, 北京今天天气怎么样)]}): print(chunk)Plan-and-SolvePlan-and-Solve 顾名思义这种范式将任务处理明确地分为两个阶段**先规划 (Plan)后执行 (Solve)**。Plan-and-Solve Prompting 由 Lei Wang 在2023年提出。其核心动机是为了解决思维链在处理多步骤、复杂问题时容易“偏离轨道”的问题。Plan-and-Solve 将整个流程解耦为两个核心阶段规划阶段 (Planning Phase) 首先智能体会接收用户的完整问题。它的第一个任务不是直接去解决问题或调用工具而是将问题分解并制定出一个清晰、分步骤的行动计划。执行阶段 (Solving Phase) 在获得完整的计划后智能体进入执行阶段。它会严格按照计划中的步骤逐一执行。每一步的执行都可能是一次独立的 LLM 调用或者是对上一步结果的加工处理直到计划中的所有步骤都完成最终得出答案。imgimport operatorfrom typing import Annotated, List, Tuple, Unionfrom typing_extensions import TypedDictfrom pydantic import BaseModel, Fieldfrom langgraph.graph import StateGraph, START, ENDfrom langchain_openai import ChatOpenAI# 1. 定义状态 (State)class PlanExecuteState(TypedDict): input: str # 原始问题 plan: List[str] # 当前待办清单 past_steps: Annotated[List[Tuple], operator.add] # 已完成的步骤和结果 response: str # 最终答案# 2. 定义结构化输出模型 (用于 Planner)class Plan(BaseModel): 步骤清单 steps: List[str] Field(description为了回答问题需要执行的步骤)# 3. 定义节点逻辑model ChatOpenAI( temperature0.6, modelglm-4.6v, openai_api_key, openai_api_basehttps://open.bigmodel.cn/api/paas/v4/)planner_model model.with_structured_output(Plan, methodfunction_calling)# --- 节点 A: 规划者 ---def planner_node(state: PlanExecuteState): plan planner_model.invoke(f针对以下问题制定计划: {state[input]}) return {plan: plan.steps}# --- 节点 B: 执行者 (这里简化了工具调用) ---def executor_node(state: PlanExecuteState): step state[plan][0] # 取当前第一步 print(f--- 正在执行: {step} ---) # 模拟工具执行结果 result f已完成 {step} 的查询结果为: [模拟数据] return {past_steps: [(step, result)], plan: state[plan][1:]}# --- 节点 C: 重规划者 (决定是继续还是结束) ---def replanner_node(state: PlanExecuteState): ifnot state[plan]: # 如果清单空了让 AI 生成最终总结 summary model.invoke( f请基于已完成的步骤和结果给出最终答案{state[past_steps]} ) return {response: summary.content} return {response: None}# 4. 路由逻辑def should_continue(state: PlanExecuteState): if state[response]: return END returnexecutor# 5. 编排图workflow StateGraph(PlanExecuteState)workflow.add_node(planner, planner_node)workflow.add_node(executor, executor_node)workflow.add_node(re-planner, replanner_node)workflow.set_entry_point(planner)workflow.add_edge(planner, executor)workflow.add_edge(executor, re-planner)# 循环逻辑根据 re-planner 的判断决定是否回 executorworkflow.add_conditional_edges(re-planner, should_continue)app workflow.compile()# 6. 测试input_query {input: 对比北京和上海的天气哪个更热}for event in app.stream(input_query): print(event)ReflectionReflection 机制的核心思想正是为智能体引入一种事后post-hoc的自我校正循环使其能够像人类一样审视自己的工作发现不足并进行迭代优化。 Reflection 框架是Shinn, Noah 在2023年提出其核心工作流程可以概括为一个简洁的三步循环执行 - 反思 - 优化。**执行 (Execution)**首先智能体使用我们熟悉的方法如 ReAct 或 Plan-and-Solve尝试完成任务生成一个初步的解决方案或行动轨迹。这可以看作是“初稿”。**反思 (Reflection)**接着智能体进入反思阶段。它会调用一个独立的、或者带有特殊提示词的大语言模型实例来扮演一个“评审员”的角色。这个“评审员”会审视第一步生成的“初稿”并从多个维度进行评估例如事实性错误是否存在与常识或已知事实相悖的内容逻辑漏洞推理过程是否存在不连贯或矛盾之处效率问题是否有更直接、更简洁的路径来完成任务遗漏信息是否忽略了问题的某些关键约束或方面 根据评估它会生成一段结构化的**反馈 (Feedback)**指出具体的问题所在和改进建议。**优化 (Refinement)**最后智能体将“初稿”和“反馈”作为新的上下文再次调用大语言模型要求它根据反馈内容对初稿进行修正生成一个更完善的“修订稿”。imgfrom typing import TypedDictfrom langchain_openai import ChatOpenAIfrom langgraph.graph import StateGraph, START, ENDclass ReflectionState(TypedDict): prompt: str draft: str critique: str final: str iteration: intllm ChatOpenAI( temperature0.6, modelglm-4.6v, openai_api_key, openai_api_basehttps://open.bigmodel.cn/api/paas/v4/,)MAX_ITERS 2def generate_draft(state: ReflectionState): msg llm.invoke(f请写一段简短答案{state[prompt]}) return {draft: msg.content, iteration: 0}def reflect_on_draft(state: ReflectionState): prompt ( 你是严格的审稿人。请指出这段答案的问题并给出改进建议。 如果没有明显问题请只输出 NO_ISSUES。\n\n f答案:\n{state[draft]} ) critique llm.invoke(prompt) print(f--- 正在执行: reflectcritique:\n {critique.content} ---) return {critique: critique.content}def revise_draft(state: ReflectionState): prompt ( 请根据以下反馈重写答案保持简短清晰\n\n f反馈:\n{state[critique]}\n\n f原答案:\n{state[draft]} ) revision llm.invoke(prompt) print(f--- 正在执行: revise原答案:\n{state[draft]}改进后\n{revision.content} ---) return {draft: revision.content, iteration: state[iteration] 1}def finalize(state: ReflectionState): return {final: state[draft]}def should_reflect(state: ReflectionState): if state[critique].strip() NO_ISSUES: returnfinalize if state[iteration] MAX_ITERS: returnfinalize returnreviseworkflow StateGraph(ReflectionState)workflow.add_node(generate, generate_draft)workflow.add_node(reflect, reflect_on_draft)workflow.add_node(revise, revise_draft)workflow.add_node(finalize, finalize)workflow.add_edge(START, generate)workflow.add_edge(generate, reflect)workflow.add_conditional_edges(reflect, should_reflect)workflow.add_edge(revise, reflect)workflow.add_edge(finalize, END)app workflow.compile()if __name__ __main__: input_data {prompt: 用三句话解释什么是 LangGraph。} result app.invoke(input_data) print(最终答案:) print(result[final])大致流程就是首先里面需要有两个角色写稿人和审稿人然后用 should_reflect 来判断是否需要重写然后用 MAX_ITERS 来限制一下最大撰写次数。[START] | v (generate_draft) | v (reflect_on_draft) -- NO_ISSUES -- (finalize) -- [END] | | iteration MAX_ITERS ----------------------- (finalize) -- [END] | -- else -- (revise_draft) -- | v (reflect_on_draft)Multi-Agent PatternMulti-Agent 模式是将复杂的任务拆解为多个专门化、独立且可协同的微服务每个服务Agent只负责一个特定的领域。因为单个 Prompt 包含太多工具和指令会导致 LLM “迷失”模型表现下降。所以通过使用doge Agent 进行职责分离不同的 Agent 可以使用不同的 Prompt、不同的模型如 GPT-4o 负责决策Llama-3 负责写代码甚至不同的工具集。比如下面的例子中一个“PM” Agent 负责拆解任务并将子任务分发给不同的“员工Workers”。imgimgfrom typing import Annotated, Literalfrom typing_extensions import TypedDictfrom langchain_openai import ChatOpenAIfrom langgraph.graph import StateGraph, START, ENDfrom langgraph.graph.message import add_messages# Multi-agent pattern: a supervisor routes work between specialist agents.llm ChatOpenAI( temperature0.4, modelglm-4.6v, openai_api_key, openai_api_basehttps://open.bigmodel.cn/api/paas/v4/,)class State(TypedDict): messages: Annotated[list, add_messages] next: str turn: intMAX_TURNS 6def _call_agent(system_prompt: str, messages: list, name: str): response llm.invoke([{role: system, content: system_prompt}] messages) return { messages: [ {role: assistant, name: name, content: response.content} ], turn: 1, }def supervisor(state: State): if state[turn] MAX_TURNS: return {next: finish} system ( You are a supervisor managing a team: researcher, writer, critic. Choose who should act next or finish. Respond with exactly one word: researcher, writer, critic, finish. ) response llm.invoke([{role: system, content: system}] state[messages]) decision response.content.strip().lower() for option in (researcher, writer, critic, finish): if option in decision: return {next: option} return {next: finish}def researcher(state: State): system ( You are a researcher. Gather key facts and constraints for the task. Be concise and list only essential points. ) return _call_agent(system, state[messages], researcher)def writer(state: State): system ( You are a writer. Produce a clear, structured response using the context. If facts are missing, note assumptions. ) return _call_agent(system, state[messages], writer)def critic(state: State): system ( You are a critic. Identify gaps, risks, or unclear parts in the draft, then suggest improvements. ) return _call_agent(system, state[messages], critic)def route_next(state: State) - Literal[researcher, writer, critic, finish]: return state[next]builder StateGraph(State)builder.add_node(supervisor, supervisor)builder.add_node(researcher, researcher)builder.add_node(writer, writer)builder.add_node(critic, critic)builder.add_edge(START, supervisor)builder.add_conditional_edges( supervisor, route_next, { researcher: researcher, writer: writer, critic: critic, finish: END, },)builder.add_edge(researcher, supervisor)builder.add_edge(writer, supervisor)builder.add_edge(critic, supervisor)app builder.compile()if __name__ __main__: user_input 创建一款广告招商的帖子 initial_state { messages: [{role: user, content: user_input}], next: supervisor, turn: 0, } for event in app.stream(initial_state): for value in event.values(): ifmessagesin value: msg value[messages][-1] name msg.get(name, assistant) print(f[{name}] {msg[content]})AI时代未来的就业机会在哪里答案就藏在大模型的浪潮里。从ChatGPT、DeepSeek等日常工具到自然语言处理、计算机视觉、多模态等核心领域技术普惠化、应用垂直化与生态开源化正催生Prompt工程师、自然语言处理、计算机视觉工程师、大模型算法工程师、AI应用产品经理等AI岗位。掌握大模型技能就是把握高薪未来。那么普通人如何抓住大模型风口AI技术的普及对个人能力提出了新的要求在AI时代持续学习和适应新技术变得尤为重要。无论是企业还是个人都需要不断更新知识体系提升与AI协作的能力以适应不断变化的工作环境。因此这里给大家整理了一份《2026最新大模型全套学习资源》包括2026最新大模型学习路线、大模型书籍、视频教程、项目实战、最新行业报告、面试题、AI产品经理入门到精通等带你从零基础入门到精通快速掌握大模型技术由于篇幅有限有需要的小伙伴可以扫码获取1. 成长路线图学习规划要学习一门新的技术作为新手一定要先学习成长路线图方向不对努力白费。这里我们为新手和想要进一步提升的专业人士准备了一份详细的学习成长路线图和规划。2. 大模型经典PDF书籍书籍和学习文档资料是学习大模型过程中必不可少的我们精选了一系列深入探讨大模型技术的书籍和学习文档它们由领域内的顶尖专家撰写内容全面、深入、详尽为你学习大模型提供坚实的理论基础。书籍含电子版PDF3. 大模型视频教程对于很多自学或者没有基础的同学来说书籍这些纯文字类的学习教材会觉得比较晦涩难以理解因此我们提供了丰富的大模型视频教程以动态、形象的方式展示技术概念帮助你更快、更轻松地掌握核心知识。4. 大模型项目实战学以致用当你的理论知识积累到一定程度就需要通过项目实战在实际操作中检验和巩固你所学到的知识同时为你找工作和职业发展打下坚实的基础。5. 大模型行业报告行业分析主要包括对不同行业的现状、趋势、问题、机会等进行系统地调研和评估以了解哪些行业更适合引入大模型的技术和应用以及在哪些方面可以发挥大模型的优势。6. 大模型面试题面试不仅是技术的较量更需要充分的准备。在你已经掌握了大模型技术之后就需要开始准备面试我们将提供精心整理的大模型面试题库涵盖当前面试中可能遇到的各种技术问题让你在面试中游刃有余。为什么大家都在学AI大模型随着AI技术的发展企业对人才的需求从“单一技术”转向 “AI行业”双背景。企业对人才的需求从“单一技术”转向 “AI行业”双背景。金融AI、制造AI、医疗AI等跨界岗位薪资涨幅达30%-50%。同时很多人面临优化裁员近期科技巨头英特尔裁员2万人传统岗位不断缩减因此转行AI势在必行这些资料有用吗这份资料由我们和鲁为民博士(北京清华大学学士和美国加州理工学院博士)共同整理现任上海殷泊信息科技CEO其创立的MoPaaS云平台获Forrester全球’强劲表现者’认证服务航天科工、国家电网等1000企业以第一作者在IEEE Transactions发表论文50篇获NASA JPL火星探测系统强化学习专利等35项中美专利。本套AI大模型课程由清华大学-加州理工双料博士、吴文俊人工智能奖得主鲁为民教授领衔研发。资料内容涵盖了从入门到进阶的各类视频教程和实战项目无论你是小白还是有些技术基础的技术人员这份资料都绝对能帮助你提升薪资待遇转行大模型岗位。大模型全套学习资料已整理打包有需要的小伙伴可以微信扫描下方CSDN官方认证二维码免费领取【保证100%免费】