GLM-4.7-Flash代码实例：LangGraph工作流编排复杂业务逻辑-育师

GLM-4.7-Flash代码实例：LangGraph工作流编排复杂业务逻辑

1. 为什么需要LangGraph来驾驭GLM-4.7-Flash

你可能已经试过直接调用GLM-4.7-Flash——输入一句话，它立刻给出专业、流畅、中文地道的回答。但当业务逻辑变复杂时，比如“先分析用户投诉邮件的情绪，再提取关键问题，接着从知识库匹配解决方案，最后生成带安抚话术的客服回复”，单次调用就力不从心了。

这不是模型能力不够，而是大模型本身不擅长状态管理、条件分支和多步骤协同。就像给一位语言大师一张白纸和一支笔，他能写出绝妙文章，但不会自动帮你拆解任务、记住中间结果、判断下一步该做什么。

LangGraph正是为解决这个问题而生：它不是另一个LLM，而是一个可编程的工作流引擎，让你像搭积木一样把GLM-4.7-Flash的能力组织成有记忆、有判断、可中断、可重试的智能体（Agent）。本文不讲抽象概念，只给你能直接复制、粘贴、运行的代码实例——从零开始，用LangGraph把GLM-4.7-Flash变成真正可用的业务助手。

2. 环境准备：三步完成本地对接

别被“LangGraph”“vLLM”这些词吓住。本镜像已为你预装全部依赖，你只需确认三件事：

2.1 确认服务已就绪

打开浏览器访问你的Web界面地址（如https://gpu-pod...-7860.web.gpu.csdn.net/），看到顶部状态栏显示🟢模型就绪，说明vLLM推理引擎已在8000端口稳定运行。

小提示：如果显示🟡“加载中”，请耐心等待约30秒，无需刷新或重启——这是模型首次加载权重的正常过程。

2.2 安装LangGraph客户端（仅需一行）

在Jupyter Notebook或终端中执行：

pip install langgraph langchain-community python-dotenv

镜像已预装langchain-core和requests，此命令仅补充LangGraph核心组件，耗时通常小于10秒。

2.3 配置API连接（安全且简洁）

创建一个.env文件，写入你的本地API地址：

echo "GLM_API_BASE=http://127.0.0.1:8000/v1" > .env

这样后续代码就能自动读取，避免硬编码——既安全，又方便迁移。

3. 实战代码：构建一个电商售后智能体

我们以真实场景切入：用户发来一条售后消息“我买的蓝牙耳机充不上电，包装盒还在，能换新吗？”，系统需自动完成四步操作：

情绪识别：判断是愤怒、焦虑还是中性；
问题归类：识别为“充电故障”；
政策匹配：查知识库，确认7天内未拆封可换新；
生成回复：输出带同理心+明确方案+行动指引的客服话术。

下面代码就是完整可运行的LangGraph工作流，每一步都附带注释说明其作用：

3.1 定义节点函数：让每个环节各司其职

import os import json from typing import Dict, Any, Optional from langchain_core.messages import HumanMessage, SystemMessage, AIMessage from langchain_core.runnables import RunnableConfig from langgraph.graph import StateGraph, END from langchain_openai import ChatOpenAI # 从环境变量读取API配置 glm_api_base = os.getenv("GLM_API_BASE", "http://127.0.0.1:8000/v1") # 初始化GLM-4.7-Flash客户端（兼容OpenAI接口） llm = ChatOpenAI( model="/root/.cache/huggingface/ZhipuAI/GLM-4.7-Flash", base_url=glm_api_base, api_key="EMPTY", # vLLM默认无需密钥 temperature=0.3, # 降低随机性，保证逻辑稳定 max_tokens=512 ) # 【节点1】情绪识别：用系统提示词约束输出格式 def detect_sentiment(state: Dict[str, Any]) -> Dict[str, Any]: user_input = state["input"] prompt = f"""你是一个专业的客服情绪分析员。 请严格按JSON格式输出，只包含一个字段'sentiment'，值为'愤怒'、'焦虑'、'中性'之一。 不要任何解释、不要额外字符。 用户消息：{user_input}""" response = llm.invoke([SystemMessage(content=prompt)]) try: result = json.loads(response.content.strip()) state["sentiment"] = result.get("sentiment", "中性") except: state["sentiment"] = "中性" return state # 【节点2】问题归类：聚焦具体故障类型 def classify_issue(state: Dict[str, Any]) -> Dict[str, Any]: user_input = state["input"] prompt = f"""请将用户问题归类为以下四类之一：'充电故障'、'连接失败'、'音质异常'、'其他硬件问题'。 只输出类别名称，不要解释。 用户消息：{user_input}""" response = llm.invoke([SystemMessage(content=prompt)]) issue = response.content.strip() state["issue_type"] = issue if issue in ["充电故障", "连接失败", "音质异常", "其他硬件问题"] else "其他硬件问题" return state # 【节点3】政策查询：模拟知识库检索（实际可替换为向量数据库） def fetch_policy(state: Dict[str, Any]) -> Dict[str, Any]: # 简化版：根据问题类型和用户描述关键词返回策略 issue = state["issue_type"] user_input = state["input"] if issue == "充电故障" and ("包装盒" in user_input or "未拆封" in user_input): policy = "7天内未拆封可免费换新" elif issue == "充电故障": policy = "提供远程指导；若无效，寄回检测后换新" else: policy = "安排工程师远程诊断" state["policy"] = policy return state # 【节点4】生成回复：融合情绪、问题、政策，输出人性化话术 def generate_response(state: Dict[str, Any]) -> Dict[str, Any]: sentiment = state["sentiment"] issue = state["issue_type"] policy = state["policy"] # 根据情绪调整语气 tone_map = { "愤怒": "非常抱歉给您带来不便，我们完全理解您的 frustration。", "焦虑": "感谢您及时反馈，我们非常重视您的使用体验。", "中性": "您好，感谢您的咨询。" } prompt = f"""你是一名资深电商客服，正在回复用户。 请结合以下信息生成一段自然、专业、有温度的回复： - 用户情绪：{sentiment} - 问题类型：{issue} - 解决方案：{policy} 要求：1) 开头用对应语气的话术；2) 明确告知处理步骤；3) 结尾提供进一步帮助入口；4) 全文不超过120字。""" response = llm.invoke([SystemMessage(content=prompt)]) state["final_response"] = response.content.strip() return state

3.2 构建工作流图：定义执行顺序与分支逻辑

# 定义状态结构（即工作流中传递的数据容器） from typing import TypedDict, Annotated from langgraph.graph.message import add_messages class GraphState(TypedDict): input: str # 原始用户输入 sentiment: str # 情绪识别结果 issue_type: str # 问题归类结果 policy: str # 匹配的售后政策 final_response: str # 最终生成的客服回复 # 创建状态图 workflow = StateGraph(GraphState) # 注册所有节点 workflow.add_node("detect_sentiment", detect_sentiment) workflow.add_node("classify_issue", classify_issue) workflow.add_node("fetch_policy", fetch_policy) workflow.add_node("generate_response", generate_response) # 设置入口点 workflow.set_entry_point("detect_sentiment") # 定义线性执行路径（无条件） workflow.add_edge("detect_sentiment", "classify_issue") workflow.add_edge("classify_issue", "fetch_policy") workflow.add_edge("fetch_policy", "generate_response") workflow.add_edge("generate_response", END) # 编译为可执行图 app = workflow.compile()

3.3 运行并验证：一次调用，全程自动化

# 模拟用户输入 user_message = "我买的蓝牙耳机充不上电，包装盒还在，能换新吗？" # 启动工作流 result = app.invoke({"input": user_message}) # 打印每一步结果，看清LangGraph如何串联LLM能力 print(" 情绪识别：", result["sentiment"]) print("🔧 问题归类：", result["issue_type"]) print("📜 匹配政策：", result["policy"]) print("\n 最终回复：") print(result["final_response"])

预期输出效果：

情绪识别： 焦虑 🔧 问题归类： 充电故障 📜 匹配政策： 7天内未拆封可免费换新 最终回复： 感谢您及时反馈，我们非常重视您的使用体验。您购买的蓝牙耳机如包装盒完好，7天内可为您免费换新。请您提供订单号，我们将立即为您安排寄送新机。

这就是LangGraph的价值：你不再需要手动拼接4次API调用、管理中间变量、处理错误重试——所有逻辑被声明式地写在图里，LangGraph自动调度、传递状态、容错重试。

4. 进阶技巧：让工作流更健壮、更实用

上面的例子是“理想路径”，真实业务中常遇到意外。LangGraph提供了优雅的应对方式，无需改写整个流程：

4.1 添加条件分支：当政策不明确时人工介入

# 在fetch_policy节点后插入分支判断 def should_esc_to_human(state: Dict[str, Any]) -> str: """判断是否需转人工客服""" if "无法确定" in state["policy"] or "请联系" in state["policy"]: return "escalate" else: return "continue" # 修改图：在fetch_policy后添加条件边 workflow.add_conditional_edges( "fetch_policy", should_esc_to_human, { "continue": "generate_response", "escalate": "notify_human_agent" # 可定义新节点发送工单 } )

4.2 加入记忆：支持多轮对话上下文

# 修改状态结构，加入历史消息 class GraphState(TypedDict): input: str messages: Annotated[list, add_messages] # LangGraph内置消息累加器 sentiment: str # ... 其他字段保持不变 # 在generate_response节点中，用messages替代单次input def generate_response(state: Dict[str, Any]) -> Dict[str, Any]: # 使用state["messages"]获取完整对话历史 # 调用llm时传入：llm.invoke(state["messages"]) pass

4.3 流式输出支持：让前端体验更丝滑

LangGraph原生支持流式响应。只需在调用时启用stream_mode：

# 替换之前的app.invoke，改为流式 for chunk in app.stream({"input": user_message}, stream_mode="values"): if "final_response" in chunk: print(chunk["final_response"], end="", flush=True)

这能让客服后台实时显示“思考中…”“正在生成…”等过渡态，大幅提升用户等待体验。

5. 部署建议：从开发到生产的关键提醒

这套代码在镜像中可直接运行，但若要投入生产，有三点必须检查：

5.1 GPU资源隔离（防干扰）

GLM-4.7-Flash在4卡并行下性能强劲，但LangGraph工作流本身不占GPU。确保：

vLLM服务独占GPU（镜像已通过CUDA_VISIBLE_DEVICES=0,1,2,3配置）；
LangGraph进程运行在CPU上（默认行为），避免与推理引擎争抢显存。

镜像已预设此隔离，你无需修改。

5.2 错误日志集中化

当某一步骤失败（如网络超时、JSON解析错误），LangGraph会抛出异常。建议在生产环境捕获并记录：

try: result = app.invoke({"input": user_message}) except Exception as e: # 记录到统一日志 with open("/var/log/glm_langgraph_error.log", "a") as f: f.write(f"[{datetime.now()}] {str(e)} | Input: {user_message}\n") result = {"final_response": "系统暂时繁忙，请稍后再试。"}

5.3 API网关层做限流

单个GLM-4.7-Flash实例可支撑约15 QPS（取决于上下文长度）。若接入高流量客服系统，务必在Nginx或云WAF层配置：

每IP每分钟请求上限（如60次）；
并发连接数限制（如50）；
自动熔断机制（错误率>5%时暂停10秒）。

镜像未内置网关，此为生产必备项，需自行部署。

6. 总结：LangGraph不是银弹，而是杠杆

GLM-4.7-Flash是一台高性能引擎，LangGraph则是那套精密的变速箱和方向盘。没有LangGraph，你只能让它直行；有了LangGraph，你才能转弯、加速、倒车、自动泊车。

本文带你走完了从环境确认、节点编写、图构建、运行验证到生产部署的全链路。你得到的不仅是一段代码，更是一种思维范式：

把业务逻辑拆解为原子节点（识别、分类、查询、生成）；
用图定义协作关系（谁先谁后、什么条件下跳转）；
让LLM专注发挥语言能力（每个节点只做一件事，且做到极致）。

下一步，你可以尝试：

把fetch_policy替换成真正的向量检索（Chroma + GLM嵌入）；
为generate_response增加多模板A/B测试；
将整个工作流封装为FastAPI接口，供企业微信机器人调用。

复杂业务逻辑的自动化，从来不是靠堆砌模型，而是靠清晰的架构设计。LangGraph，就是那个让大模型真正落地的支点。

获取更多AI镜像
想探索更多AI镜像和应用场景？访问 CSDN星图镜像广场，提供丰富的预置镜像，覆盖大模型推理、图像生成、视频生成、模型微调等多个领域，支持一键部署。

GLM-4.7-Flash代码实例：LangGraph工作流编排复杂业务逻辑