下面代码演示了如何使用 LangGraph + OpenAI 兼容接口构建一个最小但规范的多轮对话 Agent 流程，并通过阿里云 DashScope 的 Qwen3 模型进行实际推理。本质上，这是一个 LangGraph + Qwen3 的标准模板示例，适合作为后续复杂 Agent（工具调用、分支判断、循环对话）的基础骨架。

示例代码取自《LangGraph智能体设计模式与多智能体开发》第2章。

from typing import Annotated, List
from typing_extensions import TypedDict
from langgraph.graph.message import add_messages
from langchain_openai import ChatOpenAI
from langchain_core.messages import HumanMessage, AIMessage
from langgraph.graph import StateGraph, START, END
import os
# 1️⃣ 正确的状态定义（不要用 BaseModel）
class QwenState(TypedDict):
    messages: Annotated[List, add_messages]
# 2️⃣ 配置 Qwen3（OpenAI 兼容模式）
api_key = os.getenv("DASHSCOPE_API_KEY")
llm = ChatOpenAI(
    model_name="qwen3-max",
    base_url="https://dashscope.aliyuncs.com/compatible-mode/v1",
    api_key=api_key,
    temperature=0.7
)
# 3️⃣ 生成回复节点
def generate_reply(state: QwenState) -> QwenState:
    messages = state["messages"]          # List[HumanMessage | AIMessage]
    reply = llm.invoke(messages)          # AIMessage
    return {"messages": [reply]}          # ✅ 必须返回 message 对象
# 4️⃣ 构建图
graph_builder = StateGraph(QwenState)
graph_builder.add_node("generate", generate_reply)
graph_builder.add_edge(START, "generate")
graph_builder.add_edge("generate", END)
graph = graph_builder.compile()
# 5️⃣ 测试运行
if __name__ == "__main__":
    print("===== 单轮对话 =====")
    user_input1 = "你好，介绍一下 LangGraph 的特点"
    result1 = graph.invoke({
        "messages": [HumanMessage(content=user_input1)]
    })
    print(f"用户：{user_input1}")
    print(f"Agent：{result1['messages'][-1].content}\n")
    print("===== 多轮对话 =====")
    user_input2 = "刚才说的特点中，哪一点最关键？"
    result2 = graph.invoke({
        "messages": [
            HumanMessage(content=user_input1),
            AIMessage(content=result1["messages"][-1].content),
            HumanMessage(content=user_input2)
        ]
    })
    print(f"用户：{user_input2}")
    print(f"Agent：{result2['messages'][-1].content}")

上面示例代码执行结果如下：

C:\Users\xiayu\miniconda3\envs\agent\python.exe C:\Users\xiayu\PyCharmMiscProject\LangGraph-Agentic-Design-Patterns-SRC\第2章\demo_1-2.py 
===== 单轮对话 =====
用户：你好，介绍一下 LangGraph 的特点
Agent：你好！LangGraph 是由 LangChain 团队开发的一个开源库，用于构建具有状态管理和复杂控制流的基于大语言模型（LLM）的应用程序。它特别适用于需要多步骤推理、循环、条件分支或人机协作的工作流场景。
 
以下是 LangGraph 的主要特点：
 
1. **基于图的状态机（Stateful Graph）**  
   LangGraph 允许你将应用程序建模为一个有向图（Directed Graph），其中每个节点代表一个操作（如调用 LLM、工具、人工输入等），边表示状态转移。整个图维护一个共享的“状态”（state），所有节点都可以读取和更新这个状态。
 
2. **支持循环与条件逻辑**  
   与传统的线性链式流程不同，LangGraph 支持循环（例如：反复思考、自我修正）和条件分支（例如：根据 LLM 输出决定下一步操作），从而实现更复杂的推理流程，如 ReAct、Plan-and-Execute、多轮对话等。
 
3. **与 LangChain 深度集成**  
   LangGraph 构建在 LangChain 生态之上，可以无缝使用 LangChain 的工具（Tools）、记忆（Memory）、代理（Agents）和提示模板（Prompts），同时提供了更灵活的控制流能力。
 
4. **声明式与可组合性**  
   用户通过定义节点函数和边的连接规则来声明工作流，代码结构清晰、易于调试和测试。多个子图也可以组合成更大的图，支持模块化开发。
 
5. **支持异步与流式输出**  
   LangGraph 支持异步执行，并能以流式方式返回中间结果，适合构建实时响应的交互式应用（如聊天机器人、代码=生成=器等）。
 
6. **可视化与调试友好**  
   图结构天然适合可视化，LangGraph 提供了导出为 Mermaid 或其他格式的能力，便于理解复杂流程。同时，状态变更清晰可追踪，有助于调试。
 
7. **适用于高级 LLM 应用模式**  
   特别适合实现以下模式：
   - 多智能体协作（Multi-agent collaboration）
   - 自主代理（Autonomous agents）带反思机制
   - 人类-in-the-loop 工作流（如人工审核后再继续）
   - 分阶段任务分解（如先规划再执行）
 
简而言之，LangGraph 将 LLM 应用从“线性链条”升级为“状态驱动的图”，让开发者能构建更强大、更灵活、更接近真实世界复杂性的 AI 应用。
 
如果你感兴趣，我可以提供一个简单的代码示例来演示如何用 LangGraph 构建一个带循环的代理！
 
===== 多轮对话 =====
用户：刚才说的特点中，哪一点最关键？
Agent：在 LangGraph 的诸多特点中，**最关键的一点是：支持循环与条件逻辑（即复杂控制流）的能力**。
 
### 为什么这一点最关键？
 
因为这是 LangGraph **区别于传统线性链式框架（如早期 LangChain 链）的核心突破**。让我们深入解释：
 
---
 
#### 🔄 1. **真实世界的问题不是线性的**
大多数实际应用场景——比如：
- 多轮对话中的上下文修正
- 智能体反复尝试调用工具直到成功
- 自我反思（“我刚才的回答对吗？需要重做吗？”）
- 任务规划后动态调整执行路径
 
这些都**天然包含循环、分支和状态依赖**。而传统的“输入 → 处理 → 输出”链条无法表达这种逻辑。
 
---
 
#### ⚙️ 2. **循环 + 状态 = 智能体（Agent）的基础**
LangGraph 通过“图 + 共享状态 + 条件边”实现了类似有限状态机（FSM）或工作流引擎的能力。这使得你可以构建：
- **ReAct 风格的代理**：思考 → 行动 → 观察 → 再思考……
- **带退出条件的循环**：例如“最多尝试3次工具调用，失败则求助人工”
- **动态路由**：根据 LLM 输出决定走“审批流”还是“自动执行流”
 
没有循环和条件，就无法实现真正的自主推理和容错。
 
---
 
#### 🧩 3. **其他特性都建立在此基础之上**
- “状态管理”之所以有用，是因为要在循环中持久化和更新信息；
- “多智能体协作”需要节点间反复交互；
- “人类-in-the-loop”依赖条件判断是否需要暂停等待输入；
- 可视化和调试的价值，也体现在复杂控制流难以靠直觉理解时。
 
---
 
### 类比理解：
如果把 LangChain 的普通 Chain 比作“一条直线”，  
那么 LangGraph 就是“一张可以来回走、有岔路口、还能记住你去过哪儿的地图”。
 
> ✅ 所以，**支持循环与条件逻辑**不仅是 LangGraph 的技术亮点，更是它赋能高级 LLM 应用（如自主智能体）的**基石能力**。
 
如果你正在构建需要“思考-行动-反思”闭环的应用，这一点就是不可替代的关键。
 
Process finished with exit code 0

代码结构与解析

1️⃣ 状态定义（QwenState）

class QwenState(TypedDict):
    messages: Annotated[List, add_messages]

作用：

QwenState定义了 LangGraph 中整个对话流程的 共享状态
messages是对话的核心字段，用于保存所有历史消息
Annotated[List, add_messages]的含义：
- List：表示消息列表
- add_messages：LangGraph 提供的 自动合并策略
  - 每次节点返回新的 {"messages": [...]}
  - LangGraph 会自动将这些消息追加到已有消息列表中
  - 避免在多轮对话中手动维护历史

✅ 这是 LangGraph 推荐的 标准消息状态写法

✅ 不使用 Pydantic / BaseModel，而是 TypedDict，符合 LangGraph 设计理念

2️⃣ LLM 配置（Qwen3 via OpenAI 兼容接口）

llm = ChatOpenAI(
    model_name="qwen3-max",
    base_url="https://dashscope.aliyuncs.com/compatible-mode/v1",
    api_key=api_key,
    temperature=0.7
)

作用：

使用 langchain_openai.ChatOpenAI作为客户端
通过 OpenAI API 兼容模式 访问阿里云 Qwen3
关键参数说明：
- model_name="qwen3-max"：指定使用的 Qwen3 模型版本
- base_url：DashScope 提供的 OpenAI 兼容端点
- temperature=0.7：控制生成结果的随机性

✅ 这种写法让 Qwen3 完全融入 LangChain / LangGraph 生态

✅ 无需额外封装，直接复用 LangChain 的消息体系

3️⃣ 生成回复节点（generate_reply）

def generate_reply(state: QwenState) -> QwenState:
    messages = state["messages"]
    reply = llm.invoke(messages)
    return {"messages": [reply]}

作用：

这是 LangGraph 中的一个 节点函数
输入是当前状态 state，包含历史消息
执行逻辑：
1. 从 state["messages"]中取出全部消息
2. 调用 LLM（ llm.invoke）生成回复
3. 返回一个 只包含新消息的状态更新

关键点：

reply是 AIMessage对象，不是字符串
返回格式必须是 {"messages": [...]}
LangGraph 会根据 add_messages自动合并

✅ 这是 LangGraph 中 最基础、最核心的节点实现方式

4️⃣ 构建 LangGraph 图

graph_builder = StateGraph(QwenState)
graph_builder.add_node("generate", generate_reply)
graph_builder.add_edge(START, "generate")
graph_builder.add_edge("generate", END)
graph = graph_builder.compile()

作用：

定义一个线性流程：
```
START → generate → END
```
各组件含义：
- StateGraph(QwenState)：声明图的状态类型
- add_node：注册一个处理节点
- add_edge：定义节点之间的流转关系
- compile()：编译成可执行的图对象

✅ 虽然当前只是单节点，但这是后续引入：

条件分支（ add_conditional_edges）
工具调用节点
循环对话（如 ReAct、反思机制）

的 标准起点

5️⃣ 单轮对话测试

result1 = graph.invoke({
    "messages": [HumanMessage(content=user_input1)]
})

作用：

构造一条用户消息
启动 LangGraph 执行
invoke会：
1. 初始化状态
2. 执行 generate节点
3. 返回最终状态

结果中：

result1["messages"][-1].content

就是模型生成的回复。

✅ 展示了 LangGraph 中 最简单的调用方式

6️⃣ 多轮对话测试

result2 = graph.invoke({
    "messages": [
        HumanMessage(content=user_input1),
        AIMessage(content=result1["messages"][-1].content),
        HumanMessage(content=user_input2)
    ]
})

作用：

模拟真实多轮对话
手动将：
- 上一轮的 用户输入
- 上一轮的 模型回复
- 当前的新问题
一起放入 messages

LangGraph 的处理方式：

不区分“历史 / 当前输入”
所有信息都通过 messages统一传递
LLM 看到的是完整的对话上下文

✅ 这体现了 LangGraph 的一个核心理念：

对话即消息序列，状态即上下文

总结一句话

这段代码演示了：

如何用 LangGraph 构建一个基于 Qwen3 的最小对话 Agent，通过 TypedDict 管理状态，利用 add_messages 自动维护多轮对话历史，并以 OpenAI 兼容方式调用大模型完成生成任务。

它是一个 规范、简洁、可扩展的 LangGraph + Qwen3 入门模板。

ITPub博客

LangGraph多轮对话 Agent 示例解析

代码结构与解析

1️⃣ 状态定义（QwenState）

2️⃣ LLM 配置（Qwen3 via OpenAI 兼容接口）

3️⃣ 生成回复节点（generate_reply）

4️⃣ 构建 LangGraph 图

5️⃣ 单轮对话测试

6️⃣ 多轮对话测试

总结一句话

最新文章