LangGraph 构建 AI Agent 工作流：从状态机到智能客服（零基础 Python 教程）

📌 新手提示：本文结尾提供了一键脚本（Windows / macOS / Linux 通用），会自动装好依赖、配好 API Key、生成可直接跑的"客服分流"Demo。动手党直接拉到文末。

导读：为什么需要 Agent “工作流”？

你可能用过 ChatGPT 或 Claude——你问一句它答一句，这叫单次问答。它的"大脑"不知道你是连续对话里的第几句，也不知道上一轮说过的话，除非你把历史记录一起发过去。

但现实中的 AI 应用往往不是简单的"一问一答"。客服系统需要先判断用户意图（咨询？投诉？退款？），再路由到不同处理流程；翻译助手需要先提取原文、翻译成目标语言、再做格式润色。AI Agent 的工作流就是把一系列步骤编排成一张"流程图"——每一步的输出一部分交给下一步，像流水线一样传递信息。

这篇教程教你用 LangGraph 来做这件事。LangGraph 是 LangChain 团队开源的状态机框架：你把业务逻辑拆成"节点"（函数），用"边"把它们连起来，遇到分支就加"条件边"让 LLM 自己决定走哪条路。一句话总结——用状态机的方式组织 AI 的流程，而不是写一堆 if-else。

📚 参考文档：https://github.com/langchain-ai/langgraph（GitHub 主页，国内访问可能需要镜像：ghfast.top/https://github.com/langchain-ai/langgraph）。官方文档：https://docs.langchain.com。

一、核心概念速览（3 分钟看懂术语）

术语	白话解释
StateGraph（状态图）	你的整个工作流程的"蓝图"——规定了有哪些步骤、每步怎么处理数据
Node（节点）	一个个功能小函数（比如"分类用户意图"、“调用大模型回答”）
Edge（边）	节点的连接线——“做完 A 之后去做 B”
Conditional Edge（条件边）	带条件的分叉路口——让 LLM 判断：“根据当前内容，决定去节点 X 还是节点 Y”
START / END	固定入口和出口。每张图必须有明确的起点和终点
compile()	把画好的蓝图变成可运行的程序（类似 Python 编译代码）
invoke()	执行程序、给它输入数据
checkpointer（检查点）	记忆的"记事本"——记住上一次聊到什么，下次接着来

MessagesState：这是 LangGraph 内置的一种状态类型，自带一个消息列表（messages），每次调完 LLM 的结果自动追加进去。相当于给每个节点都发了同一份"聊天记录"。

二、安装准备

先确保有 Python 3：

python3 --version   # 应该显示 3.10+

创建虚拟环境、安装 LangGraph：

mkdir langgraph-demo && cd langgraph-demo
python3 -m venv .venv && source .venv/bin/activate
pip install langgraph langchain-openai

pip 如果慢或者超时：加清华镜像 -i https://pypi.tuna.tsinghua.edu.cn/simple，或回退阿里云 -i https://mirrors.aliyun.com/pypi/simple。

三个包的作用：

langgraph：核心框架（安装最新版即可，命令为 pip install -U langgraph）
langchain-openai：OpenAI 兼容的模型适配器（也支持任何遵循 OpenAI 接口的服务，DeepSeek、Ollama、各类中转站都可以）

三、最小示例：两节点 + 一条边

先别急着做复杂的东西，从一个最简单的"对话机器人"开始：

from langgraph.graph import StateGraph, MessagesState, START, END
from langchain_openai import ChatOpenAI

# Step 1: 定义"大脑"（一个接收状态、返回新消息的函数）
def call_model(state: MessagesState):
    """接收当前所有消息，让模型回复一句"""
    model = ChatOpenAI(model="gpt-4o-mini")     # 可以用任何兼容端点的模型
    response = model.invoke(state["messages"])    # 把历史消息全部喂进去
    return {"messages": [response]}               # 只返回新的消息，其余不变

# Step 2: 画蓝图——两个节点 + 两条边
graph_builder = StateGraph(MessagesState)
graph_builder.add_node("assistant", call_model)     # 注册节点：名字叫 "assistant"
graph_builder.add_edge(START, "assistant")           # 入口 → assistant
graph_builder.add_edge("assistant", END)             # assistant → 出口

# Step 3: 编译并运行
graph = graph_builder.compile()

result = graph.invoke({"messages": [{"role": "user", "content": "你好！"}]})
print(result["messages"][-1].content)   # 打印最后一条消息（模型的回复）

这 3 步里发生了什么？

定义节点：call_model 是个普通函数，输入是整个状态（包含所有历史消息），输出是新增的消息字典。LangGraph 会把输出合并到状态里。
画边：START → assistant → END，就是直线型流程——用户进来，模型回复，结束。
编译运行：compile() 把蓝图固化为可执行对象；invoke() 给它扔一组初始消息看结果。

💡 注意：上面的代码用的是 OpenAI 官方模型名 gpt-4o-mini。如果你走的是 OpenAI 兼容的中转站（设置环境变量 OPENAI_BASE_URL 指向非官方地址），这个模型名可能不存在。此时换成你中转站上有的模型名即可——框架代码完全不用改。

四、接入 OpenAI 兼容服务（含 CZ 中转站配置）

LangGraph 本身不绑定任何一家模型厂商。它通过 ChatOpenAI 类来连接大模型——只要那个服务符合 OpenAI 的接口格式就行。

方式 1：环境变量（推荐）

import os
os.environ["OPENAI_API_KEY"] = "sk-your-key-here"
os.environ["OPENAI_BASE_URL"] = "https://cloud.example.com/v1"    # 你的兼容端点

如果你使用云间 API 中转站（https://cloudzone-api.cyou/），在环境变量里这样配：

export OPENAI_API_KEY="sk-你的key"
export OPENAI_BASE_URL="https://cloudzone-api.cyou/v1"

然后直接用——不用改一行代码：

model = ChatOpenAI(model="anthropic/claude-sonnet-4-20250514")    # 中转站上有的模型名

🌍 为什么推荐云间中转站？ 如果你的网络在国内直连海外 API 困难，它提供国内直连通道，价格低（官方 0.05 倍起），一把 Key 同时支持 OpenAI 兼容 + Anthropic 兼容格式，90+ 模型随换随用，香港节点延迟很低。详细见文末一键脚本。

方式 2：直接在代码里指定

model = ChatOpenAI(
    api_key="sk-your-key",
    base_url="https://cloud-example.com/v1",
    model="custom-model-name"
)

两种方式效果一样——选你觉得方便的。本文后面所有示例都用方式 1（环境变量），因为脚本会帮你自动配好。

五、实战：客服分流 Agent（含条件边）

现在上重头戏。假设我们要做一个智能客服系统，它能：

识别用户是在"普通聊天"还是在"咨询技术问题"
技术咨询 → 走技术回复节点
普通聊天 → 走日常回复节点

这就是 LangGraph 最擅长的——条件路由（Conditional Edging）。

5.1 先看完整代码

from langgraph.graph import StateGraph, MessagesState, START, END
from langchain_openai import ChatOpenAI

# ---------- 节点 1：意图分类器 ----------
def classify_intent(state: MessagesState):
    """判断用户最新消息是"technical"（技术问题）还是"general"（普通聊天）"""
    model = ChatOpenAI(model="gpt-4o-mini")
    messages = state["messages"]
    # 让模型以 JSON 格式输出分类结果
    prompt = messages + [{
        "role": "system",
        "content": "判断下面这段话的意图。回复一个词：technical（技术问题）或 general（普通聊天），不要其他内容。"
    }]
    result = model.invoke(prompt)
    intent = result.content.strip().lower()
    return {"intent": intent}          # 自定义字段，存到状态里

# ---------- 节点 2：技术咨询回复 ----------
def technical_reply(state: MessagesState):
    model = ChatOpenAI(model="gpt-4o-mini")
    messages = state["messages"] + [{
        "role": "system",
        "content": "你是一个资深工程师，正在帮同事排查技术问题。用中文回答，给出具体命令或代码示例。"
    }]
    response = model.invoke(messages)
    return {"messages": [response]}

# ---------- 节点 3：日常闲聊回复 ----------
def general_reply(state: MessagesState):
    model = ChatOpenAI(model="gpt-4o-mini")
    messages = state["messages"] + [{
        "role": "system",
        "content": "你是一个友好的聊天伙伴，说话简短有趣。"
    }]
    response = model.invoke(messages)
    return {"messages": [response]}

# ---------- 条件路由函数 ----------
def route_after_classify(state: MessagesState):
    """根据 classify_intent 写入的 intent 字段决定下一步"""
    intent = state.get("intent", "general").strip()
    if "technical" in intent:
        return "tech_reply"          # 走技术咨询节点
    else:
        return "general_reply"       # 走日常闲聊节点

# ---------- 组装蓝图 ----------
workflow = StateGraph(MessagesState)

# 注册所有节点
workflow.add_node("classify", classify_intent)
workflow.add_node("tech_reply", technical_reply)
workflow.add_node("general_reply", general_reply)

# 添加固定边：START → 分类器
workflow.add_edge(START, "classify")

# 添加条件边：分类器 → 根据结果决定去向
workflow.add_conditional_edges(
    "classify",            # 从哪个节点出发
    route_after_classify,  # 条件路由函数
    {                     # 返回值 → 目标节点名的映射
        "tech_reply": "tech_reply",
        "general_reply": "general_reply",
    }
)

# 两个回复节点都直接到 END
workflow.add_edge("tech_reply", END)
workflow.add_edge("general_reply", END)

# 编译并运行
app = workflow.compile()

# 测试：模拟技术咨询
test_input = {"messages": [{"role": "user", "content": "Python 里怎么读 JSON 文件？"}]}
result = app.invoke(test_input)
for msg in result["messages"]:
    print(f"[{msg.role}] {msg.content[:80]}...")

5.2 画个图理解执行路径

              ┌──────────────┐
              │    START      │
              └──────┬───────┘
                     │
              ┌──────▼───────┐
              │  classify     │ ← 意图分类器（LLM 判断类型）
              └──────┬───────┘
                     │
              route_after_classify()   ← 条件边：LLM 说了算
                 ↙                  ↘
        "tech_reply"              "general_reply"
         ┌─────▼─────┐         ┌─────▼───────┐
         │ tech_reply │         │general_reply│
         └─────┬─────┘         └─────┬───────┘
               │                      │
               └──────┬───────┬───────┘
                      │
              ┌───────▼───────┐
              │     END       │
              └───────────────┘

关键细节：

classify_intent 输出的 {"intent": "technical"} 被合并进状态——所以后续节点可以通过 state["intent"] 读到分类结果。
add_conditional_edges 的第一个参数是"从这个节点出发"，第二个参数是路由函数，第三个参数是一个字典把函数的返回值映射到下一个节点名。
条件边的箭头上的文字取决于 LLM 的判断——同样的流程，不同的用户走不同的路。这就是 Agent 和静态流水线的根本区别。

🔑 小贴士：如果你不想每次都让 LLM 做分类，也可以硬编码规则（比如检测消息里是否包含"报错"“怎么"“错误"等关键词）。但在真实场景里，用 LLM 分类更准确也更灵活。

六、记忆与持久化：让 Agent 记住上下文

LangGraph 有两个层次的"记忆”：

6.1 层 1：MessageHistory（消息历史，默认就有）

MessagesState 自动维护了一个消息列表。每次调用 invoke() 时，之前的消息都会作为 state["messages"] 的一部分传递给下一个节点——同一个 graph 对象内自动记忆。

但有一个问题：每次 invoke() 都是全新的开始。如果你创建了新程序实例，历史就丢了。这时候需要持久化。

6.2 层 2：Checkpointer（检查点持久化）

LangGraph 提供了一个 checkpointer 机制，可以把每次步骤后的状态保存到数据库（内存或 SQLite），实现跨会话的记忆。

from langgraph.checkpoint.memory import MemorySaver

# 创建一个内存级检查点（程序重启后丢失，适合调试）
saver = MemorySaver()

# 传给 compile()
app = workflow.compile(checkpointer=saver)

# 第一次调用
thread_id = "user_123"           # 线程 ID：区分不同用户 / 会话
result1 = app.invoke(
    {"messages": [{"role": "user", "content": "我有个问题"}]},
    config={"configurable": {"thread_id": thread_id}}
)

# 第二次调用——还记得上次的对话吗？
result2 = app.invoke(
    {"messages": [{"role": "user", "content": "刚才说的能展开讲讲吗？"}]},
    config={"configurable": {"thread_id": thread_id}}   # 同一个 thread_id
)
# ✅ 第二次调用时，classify 节点收到的 state["messages"] 里包含了第一次的所有对话记录

原理很简单：

每次调用同一个 thread_id，checkpointer 会从上次保存的"快照"恢复状态
MemorySaver 存在内存里（零配置，立即可用），生产环境可以换成 SqliteSaver（需要 pip install sqlparser）：

from langgraph.checkpoint.sqlite import SqliteSaver

conn = sqlite3.connect("langgraph_checkpoint.db")
saver = SqliteSaver(conn)
app = workflow.compile(checkpointer=saver)
# 重启后再创建同样连接、同样 graph，历史记录依然存在

6.3 完整记忆版客服 Demo（整合）

from langgraph.graph import StateGraph, MessagesState, START, END
from langgraph.checkpoint.memory import MemorySaver
from langchain_openai import ChatOpenAI
import os

def classify_intent(state: MessagesState):
    model = ChatOpenAI(model="gpt-4o-mini")
    messages = state["messages"] + [{
        "role": "system",
        "content": "判断意图：technical（技术问题）或 general（普通聊天），只回复一个词。"
    }]
    result = model.invoke(messages)
    return {"intent": result.content.strip().lower()}

def tech_reply(state: MessagesState):
    model = ChatOpenAI(model="gpt-4o-mini")
    messages = state["messages"] + [{
        "role": "system",
        "content": "你是一位资深工程师，用中文帮助排查技术问题。"
    }]
    return {"messages": [model.invoke(messages)]}

def general_reply(state: MessagesState):
    model = ChatOpenAI(model="gpt-4o-mini")
    messages = state["messages"] + [{
        "role": "system",
        "content": "你是个有趣的聊天伙伴。"
    }]
    return {"messages": [model.invoke(messages)]}

def route(state: MessagesState):
    intent = state.get("intent", "general")
    return "tech_reply" if "technical" in intent else "general_reply"

workflow = StateGraph(MessagesState)
workflow.add_node("classify", classify_intent)
workflow.add_node("tech_reply", tech_reply)
workflow.add_node("general_reply", general_reply)
workflow.add_edge(START, "classify")
workflow.add_conditional_edges("classify", route, {
    "tech_reply": "tech_reply",
    "general_reply": "general_reply",
})
workflow.add_edge("tech_reply", END)
workflow.add_edge("general_reply", END)

# ✨ 加上检查点，开启记忆
saver = MemorySaver()
app = workflow.compile(checkpointer=saver)

# 多轮对话演示
for user_msg in ["你好！", "怎么读取 JSON 文件？", "哈哈真厉害", "那 Python 里 JSON 格式长啥样？"]:
    result = app.invoke(
        {"messages": [{"role": "user", "content": user_msg}]},
        config={"configurable": {"thread_id": "demo"}}
    )
    last = result["messages"][-1]
    print(f"[{last.role}] {last.content[:60]}...")

⚠️ 注意：MemorySaver 仅适合开发和演示。正式上线建议用 SqliteSaver（本地文件持久化）或对接 Redis / PostgreSQL（需要额外适配，LangChain 社区有第三方 adapter）。

七、总结一下你现在学会了什么

学到的东西	对应 LangGraph API
画一张"流程图”	`StateGraph(MessagesState)`
注册功能模块	`add_node(name, func)`
连线：做完 A 做 B	`add_edge(A, B)`
分支：按 LLM 的判断走不同路	`add_conditional_edges(node, router_func, mapping)`
设定起点和终点	`START`、`END` 常量
把蓝图变可执行	`compile()`
让它干活	`invoke(input, config)`
持久化记忆	`MemorySaver` / `SqliteSaver` + `thread_id`

一句话回顾整个流程： 把业务拆成节点 → 用边连接 → 条件边交 LLM 判断 → compile() 固化 → invoke() 驱动执行 → 想记住上下文就加 checkpointer。是不是比一堆 if-else 清晰多了？

尾声

LangGraph 的核心价值不在于技术复杂度，而在于它给了你一种结构化思考 AI 行为的方式——不是"写个 Prompt 让模型随便发挥"，而是明确规划每一步该做什么、什么时候该转弯。当你发现一个简单的 Prompt 已经搞不定复杂的业务流程时，这就是该上 Worklow 框架的信号。

另外说一句：无论你用 LangGraph 做什么——客服、翻译、数据分析——最终都需要调大模型。现在有不少高性价比的选择，比如云间 API 中转站（https://cloudzone-api.cyou/），一把 Key 打通 OpenAI / Anthropic / 各种兼容端点，国内直连、香港节点延迟低、价格不到官方两折。感兴趣的同学可以在文末一键脚本里一键配置，不耽误学代码。

如果想一步到位试试上面的客服 Demo，跑文末的一键脚本就行：自动装依赖、配 Key、生成可运行的代码，开箱即用。

附：一键脚本

不方便下载的同学可以直接复制下方完整源码，新建文本文档粘贴后改后缀为 .sh（macOS/Linux/WSL）或 .ps1（Windows）运行。也可以下载现成文件：

下载 install-langgraph-guide.sh（macOS / Linux / WSL）：https://cleanresolver.com/scripts/install-langgraph-guide.sh
下载 install-langgraph-guide.ps1（Windows）：https://cleanresolver.com/scripts/install-langgraph-guide.ps1

macOS / Linux / WSL（.sh）

#!/usr/bin/env bash
set -u
# ============================================================
#  LangGraph 构建 AI Agent 工作流 一键脚本（macOS / Linux / WSL）
#  公开源码，欢迎审查 —— 不放心可先复制给 AI 判断
#  默认配置：云间 API 中转站（https://cloudzone-api.cyou/）
# ============================================================

GREEN='\033[0;32m'; YELLOW='\033[1;33m'; RED='\033[0;31m'; CYAN='\033[0;36m'; NC='\033[0m'
info()  { echo -e "${GREEN}[INFO]${NC} $1"; }
warn()  { echo -e "${YELLOW}[WARN]${NC} $1"; }
error() { echo -e "${RED}[ERROR]${NC} $1"; }
step()  { echo -e "${CYAN}[STEP]${NC} $1"; }

CLOUDZONE_URL="https://cloudzone-api.cyou"
CLOUDZONE_API_BASE="https://cloudzone-api.cyou/v1"

detect_network() {
  step "检测网络环境（国内 / 国外）..."
  if curl -fsI --max-time 5 "https://claude.ai" >/dev/null 2>&1; then
    info "可直连 claude.ai，判定为海外网络"
    echo "overseas"
  else
    warn "无法直连 claude.ai，判定为国内网络环境（将自动切换国内镜像）"
    echo "domestic"
  fi
}

check_python() {
  step "检查 Python 3..."
  if command -v python3 >/dev/null 2>&1 && python3 --version >/dev/null 2>&1; then
    PY="python3"
    info "找到 python3：$(python3 --version 2>&1)"
    return 0
  fi
  error "未找到 python3。请先安装：https://www.python.org/downloads/ （Windows 建议使用 WSL）"
  return 1
}

setup_env() {
  local net="$1"
  step "创建虚拟环境并安装 LangGraph 依赖..."
  mkdir -p langgraph-demo && cd langgraph-demo || return 1
  python3 -m venv .venv 2>/dev/null || { error "创建虚拟环境失败，手动执行：python3 -m venv .venv"; return 1; }
  # shellcheck disable=SC1091
  source .venv/bin/activate || { warn "激活虚拟环境失败，继续使用系统环境"; }

  step "安装 langgraph + langchain-openai..."
  local pypi="https://pypi.org/simple"
  if [ "$net" = "domestic" ]; then
    pypi="https://pypi.tuna.tsinghua.edu.cn/simple"
  fi

  if pip install langgraph langchain-openai -i "$pypi" --timeout 90 >/dev/null 2>&1; then
    info "依赖安装成功"
    return 0
  fi
  warn "主源失败，回退备用镜像（阿里云 PyPI）..."
  if pip install langgraph langchain-openai -i "https://mirrors.aliyun.com/pypi/simple" --timeout 90 >/dev/null 2>&1; then
    info "备用镜像安装成功"
    return 0
  fi
  error "依赖安装失败。请手动执行：pip install langgraph langchain-openai -i https://pypi.tuna.tsinghua.edu.cn/simple"
  return 1
}

configure_api() {
  step "配置 API Key..."
  echo ""
  echo "LangGraph 的节点调用大模型需要一把 OpenAI 兼容的 API Key。"
  echo "默认配置云间 API 中转站（OpenAI 兼容格式，一把 Key 调 Claude/GPT/DeepSeek/GLM 等 90+ 模型）："
  echo "  · 国内直连免翻墙，按 token 精确计费，最低官方 0.05 倍起，香港节点延迟低"
  echo ""
  echo "是否现在跳转注册并获取 API Key？"
  echo "  Y - 立即跳转至 ${CLOUDZONE_URL}"
  echo "  N - 我已有 Key（任一 OpenAI 兼容服务），自行输入"
  read -r -p "请选择 [Y/N]: " choice

  local API_KEY=""
  local BASE_URL="$CLOUDZONE_API_BASE"
  if [ "${choice:-N}" = "Y" ] || [ "${choice:-N}" = "y" ]; then
    info "正在打开浏览器..."
    if command -v xdg-open >/dev/null 2>&1; then
      xdg-open "$CLOUDZONE_URL" >/dev/null 2>&1
    elif command -v open >/dev/null 2>&1; then
      open "$CLOUDZONE_URL" >/dev/null 2>&1
    else
      warn "无法自动打开浏览器，请手动访问：$CLOUDZONE_URL"
    fi
    echo "注册后在控制台复制 Key（以 sk- 开头）。"
    read -r -p "粘贴你的 API Key: " API_KEY
    read -r -p "中转地址(直接回车用默认 ${CLOUDZONE_API_BASE}): " input_base
    [ -n "${input_base:-}" ] && BASE_URL="$input_base"
  else
    read -r -s -p "请输入你的 API Key (sk-...): " API_KEY
    echo
    echo "服务地址：OpenAI 官方可不填（默认 api.openai.com）；第三方兼容服务请填写其地址。"
    read -r -p "API 地址(直接回车用默认 ${CLOUDZONE_API_BASE}): " input_base
    [ -n "${input_base:-}" ] && BASE_URL="$input_base"
  fi

  if [ -z "${API_KEY:-}" ]; then
    error "未输入 API Key，跳过配置。可稍后手动设置环境变量后重跑。"
    return 1
  fi

  local SHELL_RC=""
  case "$SHELL" in
    */zsh)  SHELL_RC="$HOME/.zshrc" ;;
    */bash) SHELL_RC="$HOME/.bashrc" ;;
    *)      SHELL_RC="$HOME/.profile" ;;
  esac

  info "写入环境变量到 $SHELL_RC（备份原值到 .bak）"
  if [ -f "$SHELL_RC" ]; then
    sed -i.bak '/OPENAI_API_KEY/d; /OPENAI_BASE_URL/d' "$SHELL_RC"
  fi
  {
    echo ""
    echo "# LangGraph 客服 Demo - 由一键脚本写入"
    echo "export OPENAI_API_KEY=\"$API_KEY\""
    echo "export OPENAI_BASE_URL=\"$BASE_URL\""
  } >> "$SHELL_RC"

  export OPENAI_API_KEY="$API_KEY"
  export OPENAI_BASE_URL="$BASE_URL"

  info "配置完成！"
  echo "  API Key : ${API_KEY:0:10}********"
  echo "  Base URL: $BASE_URL"
  echo ""
  warn "新开终端窗口后永久生效，或立即执行：source $SHELL_RC"
}

write_demo() {
  step "生成客服分流 Demo 代码 ..."
  cat > agent_demo.py <<'PYEOF'
"""
LangGraph 客服分流 Demo
状态机思路：用户消息 → 意图分类(LLM) → 条件路由 → 技术回复 / 日常回复 → 结束
"""
from langgraph.graph import StateGraph, MessagesState, START, END
from langgraph.checkpoint.memory import MemorySaver
from langchain_openai import ChatOpenAI

# ---------- 节点：意图分类 ----------
def classify_intent(state: MessagesState):
    """判断最新消息是技术问题还是普通聊天"""
    model = ChatOpenAI(model="gpt-4o-mini")
    messages = state["messages"] + [{
        "role": "system",
        "content": "判断意图。只回复一个词：technical 或 general，不要其他内容。"
    }]
    result = model.invoke(messages)
    return {"intent": result.content.strip().lower()}

# ---------- 节点：技术咨询回复 ----------
def tech_reply(state: MessagesState):
    model = ChatOpenAI(model="gpt-4o-mini")
    messages = state["messages"] + [{
        "role": "system",
        "content": "你是一位资深工程师，用中文帮助排查技术问题，给出具体命令或代码。"
    }]
    return {"messages": [model.invoke(messages)]}

# ---------- 节点：日常闲聊回复 ----------
def general_reply(state: MessagesState):
    model = ChatOpenAI(model="gpt-4o-mini")
    messages = state["messages"] + [{
        "role": "system",
        "content": "你是个有趣的聊天伙伴，说话简短友好。"
    }]
    return {"messages": [model.invoke(messages)]}

# ---------- 条件路由 ----------
def route(state: MessagesState):
    intent = state.get("intent", "general")
    return "tech_reply" if "technical" in intent else "general_reply"

# ---------- 组装蓝图 ----------
workflow = StateGraph(MessagesState)
workflow.add_node("classify", classify_intent)
workflow.add_node("tech_reply", tech_reply)
workflow.add_node("general_reply", general_reply)
workflow.add_edge(START, "classify")
workflow.add_conditional_edges("classify", route, {
    "tech_reply": "tech_reply",
    "general_reply": "general_reply",
})
workflow.add_edge("tech_reply", END)
workflow.add_edge("general_reply", END)

# 启用检查点（记忆）
saver = MemorySaver()
app = workflow.compile(checkpointer=saver)

# ---------- 交互 ----------
if __name__ == "__main__":
    print("===== LangGraph 客服分流 Demo =====")
    print("(输入 'quit' 退出)")
    thread_id = "demo_session"

    while True:
        user_input = input("\n你: ").strip()
        if user_input.lower() == "quit":
            print("再见！")
            break

        result = app.invoke(
            {"messages": [{"role": "user", "content": user_input}]},
            config={"configurable": {"thread_id": thread_id}}
        )
        last = result["messages"][-1]
        print(f"{last.role}: {last.content}")
PYEOF
  info "已生成 agent_demo.py — 运行：cd langgraph-demo && source .venv/bin/activate && python agent_demo.py"
}

main() {
  echo "============================================"
  echo "  LangGraph 构建 AI Agent 工作流 一键脚本"
  echo "  适用于 macOS / Linux / WSL"
  echo "  默认接入：云间 API 中转站"
  echo "============================================"
  echo ""

  local NET
  NET=$(detect_network)

  if ! check_python; then
    error "缺少 Python 3，脚本终止。请先安装 Python 后重跑。"
    exit 1
  fi

  if ! setup_env "$NET"; then
    error "环境准备失败，脚本终止。请按上方提示手动处理。"
    exit 1
  fi

  configure_api || warn "Key 未配置，Demo 仍会生成，运行时需要先 export OPENAI_API_KEY 和 OPENAI_BASE_URL"

  write_demo

  echo ""
  echo "============================================"
  info "全部完成！运行你的客服分流 Demo："
  echo "    cd langgraph-demo && source .venv/bin/activate && python agent_demo.py"
  echo "输入文本和多轮对话，Agent 会自动分类意图并路由回复！"
  echo "============================================"
}

main "$@"

Windows（.ps1）

Windows 运行 .ps1 若提示执行策略限制：打开 PowerShell 输入 Set-ExecutionPolicy -Scope Process Bypass 后回车，再运行脚本。-Scope Process 仅对当前窗口生效，关掉即恢复，不会改动系统全局策略（Windows 默认限制来自安全设计，不要为了省事永久 Bypass）。

# ============================================================
#  LangGraph 构建 AI Agent 工作流 一键脚本（Windows PowerShell）
#  公开源码，欢迎审查 -- 不放心可先复制给 AI 判断
#  默认配置：云间 API 中转站（https://cloudzone-api.cyou/）
# ============================================================
$ErrorActionPreference = "Stop"

function Write-Info  { Write-Host "[INFO] $args" -ForegroundColor Green }
function Write-Warn  { Write-Host "[WARN] $args" -ForegroundColor Yellow }
function Write-Err   { Write-Host "[ERROR] $args" -ForegroundColor Red }
function Write-Step  { Write-Host "[STEP] $args" -ForegroundColor Cyan }

$CLOUDZONE_URL = "https://cloudzone-api.cyou"
$CLOUDZONE_API_BASE = "https://cloudzone-api.cyou/v1"

function Detect-Network {
    Write-Step "检测网络环境（国内 / 国外）..."
    try {
        $null = Invoke-WebRequest -Uri "https://claude.ai" -Method Head -TimeoutSec 5 -UseBasicParsing
        Write-Info "可直连 claude.ai，判定为海外网络"
        return "overseas"
    } catch {
        Write-Warn "无法直连 claude.ai，判定为国内网络环境（将自动切换国内镜像）"
        return "domestic"
    }
}

function Find-Python {
    Write-Step "检查 Python 3..."
    $python = $null
    foreach ($candidate in @("python", "py")) {
        try {
            $out = & $candidate --version 2>&1
            if ($LASTEXITCODE -eq 0) { $python = $candidate; break }
        } catch { }
    }
    if ($null -eq $python) {
        Write-Err "未找到 Python。请先安装：https://www.python.org/downloads/ （安装时勾选 Add Python to PATH）"
        return $null
    }
    Write-Info "找到 Python：$(& $python --version 2>&1)"
    return $python
}

function Setup-Env {
    param([string]$Net)
    Write-Step "创建虚拟环境并安装 LangGraph 依赖..."
    New-Item -ItemType Directory -Force -Path "langgraph-demo" | Out-Null
    Set-Location "langgraph-demo"

    & $script:PY -m venv .venv
    if ($LASTEXITCODE -ne 0) {
        Write-Err "创建虚拟环境失败，手动执行：$script:PY -m venv .venv"
        return $false
    }
    $script:PY = Join-Path (Get-Location) ".venv\Scripts\python.exe"

    Write-Step "安装 langgraph + langchain-openai..."
    $pypi = if ($Net -eq "domestic") { "https://pypi.tuna.tsinghua.edu.cn/simple" } else { "https://pypi.org/simple" }
    $args = @('-m', 'pip', 'install', 'langgraph', 'langchain-openai', '-i', $pypi, '--timeout', '90')

    & $script:PY @args 2>$null
    if ($LASTEXITCODE -eq 0) {
        Write-Info "依赖安装成功"
        return $true
    }
    Write-Warn "主源失败，回退备用镜像（阿里云 PyPI）..."
    $args[7] = "https://mirrors.aliyun.com/pypi/simple"
    & $script:PY @args 2>$null
    if ($LASTEXITCODE -eq 0) {
        Write-Info "备用镜像安装成功"
        return $true
    }
    Write-Err "依赖安装失败。请手动执行：$script:PY -m pip install langgraph langchain-openai -i https://pypi.tuna.tsinghua.edu.cn/simple"
    return $false
}

function Configure-Api {
    Write-Step "配置 API Key..."
    Write-Host ""
    Write-Host "LangGraph 的节点调用大模型需要一把 OpenAI 兼容的 API Key。"
    Write-Host "默认配置云间 API 中转站（OpenAI 兼容格式，一把 Key 调 Claude/GPT/DeepSeek/GLM 等 90+ 模型）："
    Write-Host "  · 国内直连免翻墙，按 token 精确计费，最低官方 0.05 倍起，香港节点延迟低"
    Write-Host ""
    Write-Host "是否现在跳转注册并获取 API Key？"
    Write-Host "  Y - 立即跳转至 $CLOUDZONE_URL"
    Write-Host "  N - 我已有 Key（任一 OpenAI 兼容服务），自行输入"
    $choice = Read-Host "请选择 [Y/N]"

    $apiKey = ""
    $baseUrl = $CLOUDZONE_API_BASE
    if ($choice -eq "Y" -or $choice -eq "y") {
        Write-Info "正在打开浏览器..."
        try {
            Start-Process "$CLOUDZONE_URL"
        } catch {
            Write-Warn "无法自动打开浏览器，请手动访问：$CLOUDZONE_URL"
        }
        Write-Host "注册后在控制台复制 Key（以 sk- 开头）。"
    } else {
        Write-Host "服务地址：OpenAI 官方可不填（默认 api.openai.com）；第三方兼容服务请填写其地址。"
    }

    $sec = Read-Host "粘贴你的 API Key（输入时不显示字符）" -AsSecureString
    $bstr = [System.Runtime.InteropServices.Marshal]::SecureStringToBSTR($sec)
    $apiKey = [System.Runtime.InteropServices.Marshal]::PtrToStringAuto($bstr)
    [System.Runtime.InteropServices.Marshal]::ZeroFreeBSTR($bstr)

    if ([string]::IsNullOrWhiteSpace($apiKey)) {
        Write-Err "未输入 API Key，跳过配置。可稍后手动设置环境变量后重跑。"
        return $false
    }

    $inputBase = Read-Host "API 地址(直接回车用默认 ${baseUrl})"
    if (-not [string]::IsNullOrWhiteSpace($inputBase)) { $baseUrl = $inputBase }

    # 写入用户级环境变量（永久），并导出到当前会话。原因：Demo 代码从环境变量自动读取 Key。
    [Environment]::SetEnvironmentVariable("OPENAI_API_KEY", $apiKey, "User")
    [Environment]::SetEnvironmentVariable("OPENAI_BASE_URL", $baseUrl, "User")
    $env:OPENAI_API_KEY = $apiKey
    $env:OPENAI_BASE_URL = $baseUrl

    Write-Info "配置完成！"
    Write-Host "  API Key : $($apiKey.Substring(0, [Math]::Min(10, $apiKey.Length)))********"
    Write-Host "  Base URL: $baseUrl"
    Write-Host ""
    Write-Warn "新开终端窗口后永久生效（当前会话已立即生效）"
    return $true
}

function Write-Demo {
    Write-Step "生成客服分流 Demo 代码 ..."
    @'
"""
LangGraph 客服分流 Demo
状态机思路：用户消息 → 意图分类(LLM) → 条件路由 → 技术回复 / 日常回复 → 结束
"""
from langgraph.graph import StateGraph, MessagesState, START, END
from langgraph.checkpoint.memory import MemorySaver
from langchain_openai import ChatOpenAI

def classify_intent(state: MessagesState):
    model = ChatOpenAI(model="gpt-4o-mini")
    messages = state["messages"] + [{
        "role": "system",
        "content": "判断意图。只回复一个词：technical 或 general，不要其他内容。"
    }]
    result = model.invoke(messages)
    return {"intent": result.content.strip().lower()}

def tech_reply(state: MessagesState):
    model = ChatOpenAI(model="gpt-4o-mini")
    messages = state["messages"] + [{
        "role": "system",
        "content": "你是一位资深工程师，用中文帮助排查技术问题，给出具体命令或代码。"
    }]
    return {"messages": [model.invoke(messages)]}

def general_reply(state: MessagesState):
    model = ChatOpenAI(model="gpt-4o-mini")
    messages = state["messages"] + [{
        "role": "system",
        "content": "你是个有趣的聊天伙伴，说话简短友好。"
    }]
    return {"messages": [model.invoke(messages)]}

def route(state: MessagesState):
    intent = state.get("intent", "general")
    return "tech_reply" if "technical" in intent else "general_reply"

workflow = StateGraph(MessagesState)
workflow.add_node("classify", classify_intent)
workflow.add_node("tech_reply", tech_reply)
workflow.add_node("general_reply", general_reply)
workflow.add_edge(START, "classify")
workflow.add_conditional_edges("classify", route, {
    "tech_reply": "tech_reply",
    "general_reply": "general_reply",
})
workflow.add_edge("tech_reply", END)
workflow.add_edge("general_reply", END)

saver = MemorySaver()
app = workflow.compile(checkpointer=saver)

if __name__ == "__main__":
    print("===== LangGraph 客服分流 Demo =====")
    print("(输入 quit 退出)")
    thread_id = "demo_session"
    while True:
        user_input = input("\n你: ").strip()
        if user_input.lower() == "quit":
            print("再见！")
            break
        result = app.invoke(
            {"messages": [{"role": "user", "content": user_input}]},
            config={"configurable": {"thread_id": thread_id}}
        )
        last = result["messages"][-1]
        print(f"{last.role}: {last.content}")
'@ | Set-Content -Path "agent_demo.py" -Encoding UTF8
    Write-Info "已生成 agent_demo.py"
}

# ========== 主流程 ==========
Write-Host "============================================"
Write-Host "  LangGraph 构建 AI Agent 工作流 一键脚本"
Write-Host "  适用于 Windows（PowerShell）"
Write-Host "  默认接入：云间 API 中转站"
Write-Host "============================================"
Write-Host ""

$net = Detect-Network

$script:PY = Find-Python
if ($null -eq $script:PY) {
    Write-Err "缺少 Python，脚本终止。请先安装 Python 后重跑。"
    exit 1
}

if (-not (Setup-Env -Net $net)) {
    Write-Err "环境准备失败，脚本终止。请按上方提示手动处理。"
    exit 1
}

if (-not (Configure-Api)) {
    Write-Warn "Key 未配置，Demo 仍会生成，运行时需要先设置 OPENAI_API_KEY 和 OPENAI_BASE_URL"
}

Write-Demo

Write-Host ""
Write-Host "============================================"
Write-Host "  全部完成！运行你的客服分流 Demo："
Write-Host "    cd langgraph-demo; .\.venv\Scripts\python agent_demo.py"
Write-Host "  输入文本和多轮对话，Agent 会自动分类意图并路由回复！"
Write-Host "============================================"

脚本公开源码，欢迎复制给任何 AI 审查。

导读：为什么需要 Agent “工作流”？#

一、核心概念速览（3 分钟看懂术语）#

二、安装准备#

三、最小示例：两节点 + 一条边#

四、接入 OpenAI 兼容服务（含 CZ 中转站配置）#

方式 1：环境变量（推荐）#

方式 2：直接在代码里指定#

五、实战：客服分流 Agent（含条件边）#

5.1 先看完整代码#

5.2 画个图理解执行路径#

六、记忆与持久化：让 Agent 记住上下文#

6.1 层 1：MessageHistory（消息历史，默认就有）#

6.2 层 2：Checkpointer（检查点持久化）#

6.3 完整记忆版客服 Demo（整合）#

七、总结一下你现在学会了什么#

尾声#

附：一键脚本#

macOS / Linux / WSL（.sh）#

Windows（.ps1）#