☰
【AI】AI学习笔记:LangGraph入门 三大典型应用场景与代码示例及MCP、A2A与LangGraph核心对比
2026/10/17 23:32:35 网站建设 项目流程

1. 从一次“图跑不起来”说起:LangGraph 到底解决什么问题

LangGraph 是一个用来构建有状态、多步骤智能体工作流的框架。它把 LLM 调用、工具使用、逻辑判断这些动作拆成“节点”,再用“边”把它们连起来,形成一张可控制、可观测的执行图。适合谁?适合已经会写 Python、调过 OpenAI 兼容接口,但发现单纯用 while 循环 + if 判断写 Agent 越来越乱的人。

我最开始接触 LangGraph,是因为一个多步推理的小需求:用户问一句“帮我算一下 123 加 456,再判断结果是不是偶数”,用普通函数写,逻辑分支很快就缠成一团。换成 LangGraph 之后,状态在节点之间显式传递,每一步走哪条边由条件函数决定,调试时能清楚看到“现在卡在哪个节点”。这篇就按入门路径,把环境配置、图结构定义、运行验证完整走一遍,并顺带把 MCP、A2A 与 LangGraph 的定位差异讲清楚。

核心检索词先摆出来:LangGraph 入门、LangGraph 代码示例、LangGraph 与 MCP 区别、LangGraph 与 A2A 对比。这几个词基本覆盖了新手最常搜的方向。下面所有代码都可以直接复制运行,前提是先把环境准备好。

需要说明的是,LangGraph 本身不绑定某一家模型服务,它只负责编排。真正调用 LLM 的那一步,你需要一个 OpenAI 兼容的接口地址和 Key。我这边统一用 TaoToken 的接口来演示,因为它的 Base URL 和 OpenAI SDK 完全兼容,改一行配置就能跑,省去折腾环境的时间。官网入口在 https://taotoken.net/?utm_source=taotoken_aicg_blog_end&utm_medium=csdn&utm_campaign=rewrite&utm_content= ,API 地址是 https://taotoken.net/api ,注意 API 路径不带 UTM 参数。

2. 前置准备:Python 环境、依赖安装与 TaoToken Key 获取

这一节把“跑通第一个 LangGraph 应用”所需的全部前置条件列清楚。很多人卡住不是因为 LangGraph 难,而是依赖版本对不上、Key 没配好、Base URL 写错。我按顺序拆开讲。

2.1 Python 版本与虚拟环境

LangGraph 目前对 Python 版本要求是 3.9 及以上,实测 3.10 和 3.11 最稳。建议用 venv 隔离,避免和系统里其他包冲突:

python -m venv langgraph-demo source langgraph-demo/bin/activate # Windows 用 langgraph-demo\Scripts\activate python --version

确认输出是 3.10 或 3.11 即可。如果版本太低,先升级 Python,不要硬装,后面 langchain-core 会报兼容错误。

2.2 安装核心依赖

LangGraph 的包名就是 langgraph,同时需要 langchain-openai 来做模型调用。一条命令装齐:

pip install langgraph langchain-openai python-dotenv

装完可以用下面命令确认版本,写文章时我这边是 langgraph 0.2.x:

pip show langgraph | grep Version

如果后面运行报ImportError: cannot import name 'StateGraph',八成是 langgraph 版本太旧,升级到 0.2 以上即可。

2.3 获取 TaoToken Key 并写入环境变量

打开 https://taotoken.net/api-keys ,登录后创建一个 API Key,复制出来。然后项目根目录建一个.env文件:

TAOTOKEN_API_KEY=sk-你的key TAOTOKEN_BASE_URL=https://taotoken.net/api

注意 Base URL 结尾不要多加/v1,OpenAI SDK 会自动补路径。这一点很多人踩坑,写成https://taotoken.net/api/v1反而会 404。Key 只放在.env里,不要硬编码进代码,也不要提交到 Git。

2.4 模型 ID 怎么选

TaoToken 控制台的模型列表里能看到可用模型 ID,比如gpt-4o-mini、claude-3-5-sonnet这类。入门阶段建议先用便宜、响应快的模型跑通流程,确认图结构没问题,再换更强的模型。模型 ID 直接填进ChatOpenAI(model="...")里。

到这里前置就齐了:Python 环境、三个依赖包、一个 Key、一个 Base URL、一个模型 ID。接下来进入图结构定义。

3. 可复制配置:三大场景的 LangGraph 图结构定义

这一节是全文技术核心。我把多步推理、状态化对话、工具调用编排三个场景分别写成可运行的图,每个都给出完整代码和关键参数说明。所有代码共用同一套模型初始化配置,先把这个配置抽出来。

3.1 共用的模型初始化配置

新建llm_config.py:

import os from dotenv import load_dotenv from langchain_openai import ChatOpenAI load_dotenv() def get_llm(): return ChatOpenAI( model="gpt-4o-mini", # 换成你控制台里的模型 ID api_key=os.getenv("TAOTOKEN_API_KEY"), base_url=os.getenv("TAOTOKEN_BASE_URL"), # https://taotoken.net/api temperature=0, )

这里三个关键参数必须对齐:Base URL 用https://taotoken.net/api,Key 用环境变量读取,Model ID 和控制台一致。三件套缺一不可,后面所有场景都复用这个函数。

3.2 场景一:多步推理(ReAct 模式)

多步推理的核心是“思考—行动—观察”循环。下面这个图用条件边控制循环,直到拿到最终答案才走 END:

from typing import TypedDict from langgraph.graph import StateGraph, END from llm_config import get_llm class AgentState(TypedDict): question: str thought: str action: str action_input: str observation: str answer: str llm = get_llm() def llm_decision_node(state: AgentState): prompt = f"问题:{state['question']}\n已有观察:{state.get('observation', '无')}\n请判断是否需要调用计算器,只回复 need_tool 或 direct。" resp = llm.invoke(prompt).content.strip() if "need_tool" in resp: return {"action": "calculator", "action_input": state["question"]} return {"action": "respond", "answer": "无需工具,直接回答。"} def tool_node(state: AgentState): expr = state["action_input"].replace("calculate", "").strip() try: result = eval(expr) # 仅演示,生产环境请用安全解析 except Exception as e: result = f"计算失败:{e}" return {"observation": f"计算结果为 {result}"} def response_node(state: AgentState): return {"answer": f"最终答案:{state.get('observation', 'N/A')}"} workflow = StateGraph(AgentState) workflow.add_node("llm_agent", llm_decision_node) workflow.add_node("tool", tool_node) workflow.add_node("respond", response_node) workflow.set_entry_point("llm_agent") workflow.add_conditional_edges( "llm_agent", lambda x: "tool" if x.get("action") == "calculator" else "respond", {"tool": "tool", "respond": "respond"}, ) workflow.add_edge("tool", "llm_agent") workflow.add_edge("respond", END) app = workflow.compile() result = app.invoke({"question": "Please calculate 123 + 456"}) print(result["answer"])

关键点:add_conditional_edges的第二个参数是路由函数,返回的字符串必须和第三个参数字典的 key 对上,否则会报InvalidUpdateError。tool节点执行完回到llm_agent,形成循环,这就是 ReAct 的骨架。

3.3 场景二:状态化对话(带记忆的多轮)

状态化对话的重点是让 State 里保存历史消息,每轮把历史一起喂给模型:

from typing import TypedDict, Annotated from operator import add from langgraph.graph import StateGraph, END from langgraph.checkpoint.memory import MemorySaver from llm_config import get_llm class ChatState(TypedDict): messages: Annotated[list, add] llm = get_llm() def chat_node(state: ChatState): resp = llm.invoke(state["messages"]) return {"messages": [resp]} workflow = StateGraph(ChatState) workflow.add_node("chat", chat_node) workflow.set_entry_point("chat") workflow.add_edge("chat", END) memory = MemorySaver() app = workflow.compile(checkpointer=memory) config = {"configurable": {"thread_id": "user-001"}} r1 = app.invoke({"messages": [("user", "我叫小明")]}, config) r2 = app.invoke({"messages": [("user", "我叫什么?")]}, config) print(r2["messages"][-1].content)

Annotated[list, add]表示新消息追加而不是覆盖,MemorySaver配合thread_id实现跨轮记忆。同一个 thread_id 的对话会保留上下文,换一个 thread_id 就是新会话。

3.4 场景三:工具调用编排(多工具路由)

工具调用编排是把多个工具注册进图,由模型决定调哪个:

from typing import TypedDict from langgraph.graph import StateGraph, END from langchain_core.tools import tool from llm_config import get_llm @tool def add(a: int, b: int) -> int: """两数相加""" return a + b @tool def multiply(a: int, b: int) -> int: """两数相乘""" return a * b tools = [add, multiply] llm = get_llm().bind_tools(tools) class ToolState(TypedDict): messages: list def agent_node(state: ToolState): return {"messages": [llm.invoke(state["messages"])]} def tool_node(state: ToolState): last = state["messages"][-1] results = [] for call in last.tool_calls: fn = {"add": add, "multiply": multiply}[call["name"]] results.append(fn.invoke(call["args"])) return {"messages": [("tool", str(results))]} workflow = StateGraph(ToolState) workflow.add_node("agent", agent_node) workflow.add_node("tools", tool_node) workflow.set_entry_point("agent") workflow.add_conditional_edges( "agent", lambda x: "tools" if x["messages"][-1].tool_calls else END, {"tools": "tools", END: END}, ) workflow.add_edge("tools", "agent") app = workflow.compile() out = app.invoke({"messages": [("user", "3 乘 7 等于多少")]}) print(out["messages"][-1].content)

bind_tools把工具描述注入模型,模型返回tool_calls后由tool_node执行,再回到agent汇总。这就是工具调用编排的标准闭环。

4. 验证请求:跑通第一个 LangGraph 应用并确认结果

配置写完,必须实际跑一遍才算数。这一节给出验证步骤和预期输出,帮你判断到底是图的问题还是接口的问题。

4.1 分场景运行命令

把上面三段代码分别存成demo_react.py、demo_chat.py、demo_tools.py,然后逐个运行:

python demo_react.py python demo_chat.py python demo_tools.py

4.2 预期输出对照

demo_react.py正常会打印类似最终答案:计算结果为 579。demo_chat.py第二轮会打印包含“小明”的回答,说明记忆生效。demo_tools.py会打印21,说明工具路由正确。

如果输出是空字符串或者报reading choices,先别改图,去检查接口返回。可以单独写一个最小请求验证 Key 和 Base URL:

from llm_config import get_llm print(get_llm().invoke("你好").content)

这行能正常返回,说明三件套没问题,问题在图逻辑;这行报错,问题在配置。

4.3 用模型对话页快速验证模型可用性

不想写代码验证时,可以直接打开 https://taotoken.net/models?utm_source=taotoken_aicg_blog_end&utm_medium=csdn&utm_campaign=rewrite&utm_content= ,在模型对话页里选同一个模型 ID 发一句话。能正常回复,说明模型和 Key 都没问题,再回去跑 LangGraph 就排除了接口因素。这一步能省掉大量“到底是图错还是 Key 错”的纠结。

4.4 观察图执行过程

LangGraph 支持流式观察每个节点的输出,调试时很有用:

for event in app.stream({"question": "Please calculate 123 + 456"}): print(event)

它会按节点顺序打印状态变化,你能清楚看到llm_agent先输出什么、tool又输出什么。多步推理卡住时,这个输出比 print 大法高效得多。

5. 本篇常见错排查:401、local proxy failed、reading choices、OAuth

这一节按真实报错逐条对照。这些错误我基本都遇到过,按顺序排查能覆盖九成问题。

5.1 401 Unauthorized

最常见。原因通常是 Key 没读到或写错。检查.env里变量名和代码里os.getenv的名字是否一致,注意大小写。另外确认 Key 没有多余空格,复制时容易带上换行。如果用的是系统环境变量而不是.env,确认load_dotenv()在读取之前调用。

5.2 local proxy failed / connection error

这个报错说明请求根本没发出去,通常是 Base URL 写错或网络层配置问题。确认TAOTOKEN_BASE_URL是https://taotoken.net/api,不要带/v1,也不要带结尾斜杠。如果公司网络有额外限制,换一个网络环境再试。注意不要在任何配置里写代理相关的东西,保持直连即可。

5.3 reading choices / 'str' object has no attribute 'choices'

这个报错一般出现在你把模型返回当字符串处理,但代码期望的是对象。检查llm.invoke()的返回值,.content才是文本。如果用了bind_tools,返回的是 AIMessage,取tool_calls而不是.content。另外确认 langchain-openai 版本和 langgraph 匹配,版本错位也会导致返回结构异常。

5.4 OAuth / authentication 相关报错

如果你在 Claude Code 或 Codex 这类工具里配置,报 OAuth 错误通常是认证方式选错了。这类工具要的是 API Key 模式,不是 OAuth 登录。配置时把 Base URL 填https://taotoken.net/api,Key 填 TaoToken 的 Key,Model ID 填控制台里的模型名,三件套对齐就不会走 OAuth 流程。CC Switch、Cline MCP、Codex 的auth.json都是同一个逻辑:Base URL + Key + Model ID 三件套写全,缺一个就会回退到默认认证方式从而报错。

5.5 InvalidUpdateError / 边路由不匹配

这是图逻辑错误,不是接口错误。检查add_conditional_edges里路由函数返回的字符串,是否和映射字典的 key 完全一致。大小写、下划线都要对上。另外确认set_entry_point设置的节点名和add_node注册的名字一致。

5.6 循环不退出

多步推理场景里,如果路由函数永远返回tool,图会一直循环。加一个最大步数保护:

app = workflow.compile() result = app.invoke({"question": "..."}, {"recursion_limit": 10})

recursion_limit超了会抛异常,比死循环好排查。

6. MCP、A2A 与 LangGraph 的核心对比与选型建议

跑通三个场景后,回到标题里的对比问题。这三个概念经常被混在一起,其实定位完全不同。

MCP 是模型上下文协议,解决的是“模型怎么标准化地调用外部工具和数据源”。它定义的是 Host、Client、Server、Tools、Resources 这套接口规范。你可以把它理解成智能体的“手和脚”的接入标准。LangGraph 里的某个工具节点,完全可以通过 MCP Client 去调用 MCP Server 提供的工具,两者是配合关系。

A2A 是智能体间协作协议,解决的是“多个智能体之间怎么发现、通信、分工”。它定义的是技能清单、任务委托、状态同步这些规则。你可以把它理解成多智能体团队的“组织架构和协作章程”。LangGraph 则是实现 A2A 协作模式的底层引擎之一,你用 LangGraph 的图来定义智能体之间的路由、对话和任务传递逻辑。

LangGraph 本身是工作流编排框架,解决的是“具体怎么实现一个有状态、多步骤的智能体应用”。它提供状态、节点、边、检查点这些原语。MCP 是协议层,A2A 是协作层,LangGraph 是实现层。三者互补:用 LangGraph 做核心引擎,按 A2A 的协作理念设计多智能体图,图中每个智能体需要行动时通过 MCP 调用外部工具。

选型上给个直接建议:单智能体、多步推理、需要状态管理,直接用 LangGraph;需要接入大量外部工具且希望标准化,叠加 MCP;需要多个专业智能体分工协作,参考 A2A 的设计思路再用 LangGraph 落地。入门阶段先把 LangGraph 的三个场景跑通,再往上叠 MCP 和 A2A,路径最顺。

如果你打算长期做编码类 Agent,或者要把这套图跑在持续任务上,可以看下 Coding Plan:https://taotoken.net/coding-plan?utm_source=taotoken_aicg_blog_end&utm_medium=csdn&utm_campaign=rewrite&utm_content= 。接入文档在 https://taotoken.net/doc?utm_source=taotoken_aicg_blog_end&utm_medium=csdn&utm_campaign=rewrite&utm_content= ,遇到配置问题对照文档里的 Base URL 和 Key 说明最快。控制台在 https://taotoken.net/console?utm_source=taotoken_aicg_blog_end&utm_medium=csdn&utm_campaign=rewrite&utm_content= ,模型 ID 和用量都在里面看。

最后留一个我踩过的坑:MemorySaver是内存存储,进程重启记忆就没了。要持久化得换SqliteSaver或接数据库。入门阶段先用内存版跑通逻辑,确认图没问题再换存储,能少走很多弯路。

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

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

立即咨询