LangGraph 核心概念1.0 先看个最小例子from typing_extensions import TypedDictfrom langgraph.graph import StateGraph, START, ENDclassCountState(TypedDict): count: intdefincrement(state: CountState) - dict: return {count: state[count] 1}defshould_stop(state: CountState) - str: returnstopif state[count] 3elsecontinuegraph StateGraph(CountState)graph.add_node(increment, increment)graph.add_edge(START, increment)graph.add_conditional_edges( increment, should_stop, {continue: increment, stop: END},)app graph.compile()print(app.invoke({count: 0})) # {count: 3}LangGraph 的 API解读:StateGraph(CountState)—— 一张图,全局 state 是CountStateadd_node(increment, increment)—— 注册节点add_edge(START, increment)—— 普通边add_conditional_edges(..., should_stop, {...})—— 条件边compile()—— 把声明式结构变成可执行对象invoke(...)—— 启动图,跑一次到结束条件边:“stop” → END, “continue” → 回 increment 自身。1.1 图的三个核心元素┌───────────────────────────────────────────┐│ ││ State → 全局上下文 / 状态机的全局变量 ││ Node → 处理函数 / 读 state 写回 state ││ Edge → 节点之间的转移规则 ││ │└───────────────────────────────────────────┘LangGraph 对应到代码上,就是StateGraph类的add_node/add_edge/ 状态 schema 定义。1.2 State 在节点间自动累积节点之间传 state,要回答一个问题:节点返回的部分 state 是覆盖还是追加?默认是覆盖。简单字段(比如count: int)这样没问题,但messages: list就不行——每一轮 LLM 调用、工具调用都要把新消息追加到对话历史里。LangGraph 用类型注解里挂一个如何合并的小函数来解决:from collections.abc import Sequencefrom typing import Annotatedfrom typing_extensions import TypedDictfrom langchain_core.messages import BaseMessagefrom langgraph.graph.message import add_messagesclass AgentState(TypedDict): messages: Annotated[Sequence[BaseMessage], add_messages]add_messages就是那个合并函数,合并规则是追加而不是替换:# 节点 A 返回 {messages: [msg1, msg2]} → state[messages] 追加 msg1, msg2# 节点 B 返回 {messages: [msg3]} → state[messages] 追加 msg3这样 ReAct 那张图才能循环跑 N 轮,每一轮的对话历史都保留下来,下一轮 LLM 看到的是完整历史。不挂合并函数会怎样:class AgentState(TypedDict): messages: list # ← 默认行为 替换# 节点返回 {messages: [AIMessage(content...)]}# state[messages] 整个被替换为那条 AI 消息# HumanMessage(天气?) 丢了,下一轮 LLM 不知道用户问的什么1.3 边的两类普通边add_edge(A, B)—— A 跑完一定跳 B条件边add_conditional_edges(A, router_fn, path_map)—— A 跑完调router_fn(state),根据返回值查path_map决定下一步ReAct 的要不要继续调工具循环,天然需要条件边。1.4 入口与终止入口:START节点,或set_entry_point(agent)显式指定第一个节点终止:END特殊标识,代表图跑完了LangGraph 内置了START哨兵节点(不需要add_node(START, ...)注册),compile()后图从它开始执行。如何用 LangGraph 实现 ReAct 模式2.1 ReAct 是什么ReAct深度解析从方法论到工程实践ReAct(Reason-Action)的核心:让大模型在思考的同时也能动手——不是先全部想完再调工具,而是每想一步都可能动手,再根据工具返回的结果继续推理。循环路径:思考 → 行动 → 观察 → 思考 → … 退出路径:思考 → 最终回答模型自己决定每一步该做什么。这种思考 → 行动 → 观察 → 思考 → …循环可能跑很多次才停,也可能一轮就够。2.2 用 LangGraph 中图的概念实现 ReAct把 ReAct 翻译成 LangGraph 的图,3 个决策:决策 1:有几个节点?—— 2 个。ReAct 行为里两个不同的工作:调大模型让它思考、跑工具拿到结果。agent节点 调大模型tools节点 执行模型声明的工具调用决策 2:有几条边?—— 1 条普通边 1 条条件边。tools跑完一定回到agent→ 普通边tools → agentagent跑完后不一定去tools,模型觉得答完了就直接结束 → 条件边agent → {tools, END}决策 3:条件边的路由器怎么写?—— 是否继续等价于:最近一条 AI 消息里有没有tool_calls?有 → 模型想调工具,跳tools没有 → 模型给出最终答案,跳END完整拓扑:决策点放在agent之后(不是tools之后),因为要不要继续取决于 AI 的最新输出,不是工具结果。回边只画tools → agent,反过来不会发生。2.3 核心代码解读2.3.1 图装配build_app把图怎么装起来全说清楚:state schema、三个处理单元、Provider/checkpointer 全部串成一张可执行图。先看完整骨架,后面 §2.3.2-§2.3.4 再拆每个零件。from langgraph.checkpoint.memory import InMemorySaverfrom langgraph.graph import END, StateGraphfrom langgraph.graph.state import CompiledStateGraphdefbuild_app(modelNone, toolsNone, checkpointerNone) - CompiledStateGraph: if tools isNone: tools ALL_TOOLS # §2.3.3 ② 默认工具集 if model isNone: model make_default_model().bind_tools(tools) # §2.3.4 ② ③ checkpointer checkpointer or InMemorySaver() workflow StateGraph(AgentState) # §2.3.2 workflow.add_node(agent, make_call_model(model)) # §2.3.3 ① workflow.add_node(tools, make_tool_node(tools)) # §2.3.3 ② # 入口 workflow.set_entry_point(agent) # 条件边:agent 出来后路由器判下一步 workflow.add_conditional_edges( agent, should_continue, # §2.3.3 ③ {continue: tools, end: END}, ) # 回边 workflow.add_edge(tools, agent) # 编译成可执行对象 return workflow.compile(checkpointercheckpointer)add_conditional_edges是三元组:源节点 路由器 路径映射表。set_entry_point(agent)是add_edge(START, agent)的简写。compile(checkpointer...)实例化整张图,带上InMemorySaver—— 这个 checkpointer不是摆设,它给图配上一个按thread_id持久化 state的存储后端。2.3.2 状态 schemafrom collections.abc import Sequencefrom typing import Annotatedfrom typing_extensions import TypedDictfrom langchain_core.messages import BaseMessagefrom langgraph.graph.message import add_messagesclass AgentState(TypedDict): ReAct 唯一的全局状态:messages 列表。 messages: Annotated[Sequence[BaseMessage], add_messages]AgentState是图的全局 state,只有 1 个字段messages,装着对话历史。每节点产生的新消息按add_messages的合并规则追加到这个列表里。2.3.3 三个处理单元① 大模型调用(agent节点)from langchain_core.messages import SystemMessagefrom langchain_core.runnables import RunnableConfigSYSTEM_PROMPT SystemMessage( You are a helpful AI assistant, please respond to the users query to the best of your ability!)defmake_call_model(model): 工厂:返回一个能被 LangGraph 当作节点调用的函数。 defcall_model(state: AgentState, config: RunnableConfig None) - dict: response model.invoke([SYSTEM_PROMPT] list(state[messages]), config) return {messages: [response]} # list 包一层,因为合并函数需要列表 return call_model工厂模式是为了测试时方便换底层模型——FakeMessagesListChatModel是langchain_core自带的假模型,不需要调真 API:fake_model FakeMessagesListChatModel(responses[...])call_model make_call_model(fake_model)graph.add_node(agent, call_model)直接定义call_model(state)会硬编码(hardcode)用哪个模型,测试得monkeypatch替换。工厂把用什么模型作为参数注入。② 工具执行(tools节点)import jsonfrom langchain_core.messages import ToolMessagedefmake_tool_node(tools): tools_by_name {t.name: t for t in tools} # 只算一次,后续节点调用复用 deftool_node(state: AgentState) - dict: outputs [] # 取最新那条 AI 消息里的所有 tool_calls,逐个执行 for tool_call in state[messages][-1].tool_calls: tool_result tools_by_name[tool_call[name]].invoke(tool_call[args]) outputs.append( ToolMessage( contentjson.dumps(tool_result), nametool_call[name], tool_call_idtool_call[id], # 必须跟 AI 消息的 id 配对 ) ) return {messages: outputs} return tool_nodestate[messages][-1].tool_calls—— 只读最新那条 AI 消息里的 tool_calls。LLM 一次能 emit 多个 tool_call,逐个跑。tool_call_idtool_call[id]—— 协议强约束。每条 ToolMessage 必须带tool_call_id跟产生它的 AI 消息里tool_call[id]配对,否则下一轮 LLM 看不到对应关系。工具定义from langchain_core.tools import tooltooldefget_weather(location: str) - str: Call to get the weather from a specific location. ifany(city in location.lower() for city in [sf, san francisco]): returnIts sunny in San Francisco, but you better look out if youre a Gemini . returnfI am not sure what the weather is in {location}ALL_TOOLS [get_weather]tool装饰器把普通函数转成BaseTool,bind_tools(tools)就会把它的函数签名 docstring(nameget_weather、参数location: str、docstring 当 description)打包成 JSON Schema 塞进 LLM 调用。注:项目的get_weather是占位实现(notebook 作者标注 “Don’t let the LLM know this though ”),真实场景要接天气 API。这里只演示工具协议怎么走通。③ 是否继续(条件边路由器)def should_continue(state: AgentState) - str: 最近一条 AI 消息有没有 tool_calls。 last_message state[messages][-1] if not last_message.tool_calls: return end return continue把 ReAct 论文里模型在 Thought 字段里写Finish[answer]“这种文本级约定,翻译成了协议级判定——只要消息没有tool_calls,就等于答完了”。模型不用老老实实写Finish[...]。2.3.4 LLM Provider 与模型工厂① Provider 抽象LLM 服务都长这样:模型名 base_url api key 三件套。把它们收成一张表,加 provider 就加一行:from dataclasses import dataclassdataclass(frozenTrue)classProvider: key_env: str # api key 存在哪个 env var model_default: str # 默认模型名 base_url_default: str | None# 默认 base_url(None 走 ChatOpenAI 官方)PROVIDERS: dict[str, Provider] { zhipu: Provider( key_envZHIPUAI_API_KEY, model_defaultGLM-5.1, base_url_defaulthttps://open.bigmodel.cn/api/coding/paas/v4, ), minimax: Provider( key_envMINIMAX_API_KEY, model_defaultMinMax-M3, base_url_defaulthttps://api.minimaxi.com/v1, ), openai: Provider( key_envOPENAI_API_KEY, model_defaultgpt-4o-mini, base_url_defaultNone, ),}三家都走 OpenAI 兼容的/chat/completions接口 OpenAI 风格tools协议——所以只要换 base_url 和 api key,LangGraph 这边的图一行业务代码都不用改。② 工厂方法make_default_modelimport osimport sysfrom langchain_openai import ChatOpenAIDEFAULT_PROVIDER zhipudefrequire_env(name: str) - str: 读必需 env var,缺失就 stderr sys.exit(2)(用法错误惯例)。 val os.environ.get(name) ifnot val: sys.stderr.write( ferror: required env var {name} is not set. fCopy .env.example to .env and fill it in, or export it fin your shell.\n ) sys.exit(2) return valdefmake_default_model() - ChatOpenAI: provider os.getenv(LLM_PROVIDER, DEFAULT_PROVIDER).lower() if provider notin PROVIDERS: sys.stderr.write( ferror: unknown LLM_PROVIDER{provider}. fChoose one of: {, .join(sorted(PROVIDERS))}.\n ) sys.exit(2) cfg PROVIDERS[provider] api_key require_env(cfg.key_env) kwargs {model: cfg.model_default, api_key: api_key} if cfg.base_url_default: kwargs[base_url] cfg.base_url_default return ChatOpenAI(**kwargs)ChatOpenAI是langchain_openai封装的 OpenAI 客户端,base_url一改,就从 OpenAI 切到 GLM / MiniMax / DeepSeek 等任意兼容端点。③bind_tools(tools)— 让模型知道有工具model make_default_model()model_with_tools model.bind_tools(tools) # ← 关键一行bind_tools把tools列表里每个工具的 schema(name、参数、docstring)塞进 LLM 调用,协议级告诉模型:“你可以 emittool_calls字段”。模型返回的AIMessage因此有两种形态:模型觉得要调工具模型觉得答完了AIMessage(content, tool_calls[...])AIMessage(contentIts sunny...)——这就是路由器should_continue工作的前提bind_tools装上协议,should_continue才有tool_calls字段可读。2.3.5 REPL 入口import uuidfrom collections.abc import Iterablefrom langchain_core.messages import BaseMessagefrom .config import load_envfrom .graph import build_appBANNER ( ReAct REPL — type a message and press Enter.\n :clear reset the conversation history (new thread)\n :quit exit the REPL\n Ctrl-D exit the REPL)def_format_message(msg: BaseMessage | tuple) - str: LangGraph stream_modevalues 的元素可能是 BaseMessage 也可能是历史 tuple,两种都格式化输出。 ifisinstance(msg, tuple): returnstr(msg) return msg.pretty_print()defprint_stream(stream: Iterable[dict]) - None: 每个 state 快照只打最后一条新追加的消息,避免 spam。 for s in stream: msg s[messages][-1] print(_format_message(msg))defrepl() - None: load_env() # 读 .env app build_app() # §2.3.1 装好的图 print(BANNER) thread_id str(uuid.uuid4()) config {configurable: {thread_id: thread_id}} # ① 运行时配置 whileTrue: try: user_input input(\n ).strip() except (EOFError, KeyboardInterrupt): print(\nbye) return ifnot user_input: continue if user_input :quit: print(bye) return if user_input :clear: # ③ 换 thread_id thread_id str(uuid.uuid4()) config {configurable: {thread_id: thread_id}} print((history cleared)) continue inputs {messages: [(user, user_input)]} # ② 初始 state print_stream(app.stream(inputs, config, stream_modevalues))if __name__ __main__: repl()①config—— 运行时配置config {configurable: {thread_id: thread_id}}configurable是 LangGraph 预留的 key,图运行时从这里读每次调用可变的参数。关键是thread_id:LangGraph 的 checkpointer 按这个值存/读 state——换thread_id 开新对话,同一个 继续上次对话。configurable里还能塞别的(user_id/recursion_limit等),本教程只用thread_id。②inputs—— 图的初始 stateinputs {messages: [(user, user_input)]}字段名要跟AgentState对得上——这里只有messages,所以inputs只放messages。元组(user, ...)是langchain_core简写,langchain_core看到元组就按第一个元素当 role 转成对应BaseMessage(user→HumanMessage、assistant→AIMessage)。直接传HumanMessage(content...)也行,只是元组写法更简洁。③:clear重置 thread_idif user_input :clear: thread_id str(uuid.uuid4()) config {configurable: {thread_id: thread_id}}:clear不删 state(checkpointer 不暴露删接口)——而是换一个新 thread_id,旧 thread 还在内存里,新 thread 从空 state 启动。InMemorySaver进程退出就清空,所以对用户来说删等价于换。2.3.6 一轮 ReAct 跑起来输入:用户问 “what’s the weather in sf?”,state[messages] [HumanMessage(whats the weather in sf?)]。Agent 节点的输出 ≈ ReAct 的思考 行动,Tools 节点的输出 ≈ ReAct 的观察,should\_continue路由 ≈ 下一步该思考还是该停止决策,add\_messages合并规则 ≈ 把每一轮的输出串成完整历史。一轮跑完 3 个节点、2 段边(条件边 回边),ReAct 调一次工具就给出最终答案。总结LangGraph用 Python 类表达图的语义。核心元素:State(全局上下文) Node(处理函数) Edge(转移规则)。State 自动在节点间累积,关键靠Annotated[..., 合并函数]的类型注解装配合并规则,最常见是add_messages(追加而不是覆盖)。边的两类:普通边无条件,条件边靠路由器函数决定下一步ReAct在 LangGraph 里翻译成2 节点 1 条件边:节点agent调大模型,节点tools跑工具,条件边agent → {tools, END}由should_continue路由,普通回边tools → agentReAct 循环不是for/while,是节点之间反复跳。决策点放在agent之后,因为要不要继续取决于 AI 的最新输出should_continue把 ReAct 论文里Thought 里写Finish[answer]的文本约定,改成协议级判定(tool_calls字段是否存在)ToolMessage 的tool_call_id必须跟 AI 消息里tool_call.id配对,这是 OpenAI/Anthropic/GLM 的协议强约束理解这套2 节点 1 条件边的最小骨架,后面所有 LangGraph agent 模式都只是在这套骨架上增加节点、增加状态字段、调整终止策略:把END换成ask_human(human-in-the-loop)给agent节点配更复杂的 prompt,产出结构化输出多加几个agent节点,分别负责规划和执行在节点之间加 self-critique 节点本质上都是图的形态变化,LangGraph 的基本元素没变。学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%免费】