LangGraph实战
·
LangGraph实战
一、LangGraph简介
1、LangGraph特性
- 持久化执行，任何故障后皆可无缝恢复
- 完整的内存管理，整合短期记忆和长期记忆
- 生产级可靠性
- 可观测性，与LangSmith集成
一个为生产环境而生的低层级编排框架
- 完全控制，可直接编写prompt
- 显示状态，State, agent运转的状态记录
- 图状结构
- 生产级可靠
2、核心概念
2.1 state
单一事实来源
所有数据通过state在节点间传递和更新
2.2 node
每个node负责一个单一的职责
一般执行LLM调用、工具调用
2.3 Edges
node间的连接线
普通边：实线
条件边：虚线
3、适用场景
- 多步复杂推理
- 长时间运行任务
- 需要人的介入
- 追求生产级稳定性
二、LangGraph代码示例
1、基本Hello
安装依赖
> pip install langgraph langchain langchain-openai pygraphviz
代码示例：
from typing import TypedDict
from langgraph.graph import StateGraph, START, END
# 定义状态TestState
class TestState(TypedDict):
username: str
# 定义节点hello和hook
def hello(state: TestState) -> dict:
print(f"hello {state['username']}")
return {}
def hook(state: TestState) -> dict:
print("握手")
return {}
# 创建状态图
graph = StateGraph(TestState)
# 添加节点node
graph.add_node("hello", hello)
graph.add_node("hook", hook)
# 添加边
graph.add_edge(START, "hello")
graph.add_edge("hello", "hook")
graph.add_edge("hook", END)
# 编译状态图
app = graph.compile()
# 创建输入参数
username = "martin"
# 调用状态图
result = app.invoke({"username": username})
# 获取流程图
mermaid_str = app.get_graph().draw_mermaid()
html_content = f"""<!DOCTYPE html>
<html>
<head>
<meta charset="utf-8">
<title>LangGraph Flow</title>
</head>
<body>
<pre class="mermaid">{mermaid_str}</pre>
<script src="https://cdn.jsdelivr.net/npm/mermaid/dist/mermaid.min.js"></script>
<script>mermaid.initialize({{startOnLoad:true}});</script>
</body>
</html>"""
output_path = r"e:\langGraphTest\001demo\flowchart.html"
with open(output_path, "w", encoding="utf-8") as f:
f.write(html_content)
print(f"\n流程图已保存到: {output_path}")
print("请在浏览器中打开该文件查看流程图")
运行代码产生的流程图:
2、条件分支示例
from typing import TypedDict
from langgraph.graph import StateGraph, START, END
class WeatherState(TypedDict):
temperature: int
recommendation: str
# 为温度赋值，方便后续测试
def check_temperature(state: WeatherState) -> dict:
return {"temperature": 25}
def hot(state: WeatherState) -> dict:
print("推荐: 防晒")
return {"recommendation": "防晒"}
def warm(state: WeatherState) -> dict:
print("推荐: 半袖")
return {"recommendation": "半袖"}
def cold(state: WeatherState) -> dict:
print("推荐: 外套")
return {"recommendation": "外套"}
# 从state中获取上面的赋值，根据温度返回不同的路由
def route_by_temperature(state: WeatherState) -> str:
temp = state["temperature"]
if temp >= 26:
return "hot"
elif temp >= 20:
return "warm"
else:
return "cold"
graph = StateGraph(WeatherState)
graph.add_node("check", check_temperature)
graph.add_node("hot", hot)
graph.add_node("warm", warm)
graph.add_node("cold", cold)
# 添加边，从START到check，然后根据温度返回不同的路由
graph.add_edge(START, "check")
graph.add_conditional_edges(
"check",
route_by_temperature,
{
"hot": "hot",
"warm": "warm",
"cold": "cold",
}
)
graph.add_edge("hot", END)
graph.add_edge("warm", END)
graph.add_edge("cold", END)
app = graph.compile()
result = app.invoke({})
print(f"\n当前温度: {result['temperature']}°C, 最终推荐: {result['recommendation']}")
mermaid_str = app.get_graph().draw_mermaid()
html_content = f"""<!DOCTYPE html>
<html>
<head>
<meta charset="utf-8">
<title>LangGraph Conditional Branching</title>
</head>
<body>
<pre class="mermaid">{mermaid_str}</pre>
<script src="https://cdn.jsdelivr.net/npm/mermaid/dist/mermaid.min.js"></script>
<script>mermaid.initialize({{startOnLoad:true}});</script>
</body>
</html>"""
output_path = r"e:\langGraphTest\001demo\flowchart_conditional.html"
with open(output_path, "w", encoding="utf-8") as f:
f.write(html_content)
print(f"\n流程图已保存到: {output_path}")
print("请在浏览器中打开该文件查看流程图")
运行后的流程图:
三、LangGraph实例
1、功能说明
这是一个LangGraph实例。作用是对用户的输入通过调用LLM对用户的输入进行智能诊断，诊断后返回用户输入中的敏感词汇和评级。再将诊断的结果交给LLM，根据诊断结果判断用户输入是否需要人工审核、还是通过检测。
# =============================================================================
# 内容智能检测 Agent
# 功能：调用 DeepSeek LLM 对用户输入进行脏话检测（脏话），
# 并根据检测结果由 LLM 判断是否需要人工审核或直接通过
# 流程：START → detect(检测) → analyze(分析决策) → human_review / pass → END
# =============================================================================
import json
import urllib.request
from typing import TypedDict
from langgraph.graph import StateGraph, START, END
# ========================= 配置区域 =========================
# DeepSeek API 配置
API_KEY = "sk-key"
API_URL = "https://api.deepseek.com/chat/completions"
MODEL = "deepseek-v4-flash"
# ========================= 状态定义 =========================
# 定义 LangGraph 工作流的状态结构，所有节点共享并读写此状态
class ContentState(TypedDict):
user_input: str # 用户原始输入
detection_result: str # LLM 检测的原始返回结果
sensitive_words: str # 检测到的敏感词（逗号分隔）
reason: str # 检测原因 + 审核决策原因
risk_level: str # 风险评级：低 / 中 / 高
need_review: bool # 是否需要人工审核（True/False）
review_decision: str # 最终审核决策："需要人工审核" 或 "通过"
final_result: str # 格式化后的最终输出结果
# ========================= 工具函数 =========================
def call_deepseek(messages: list) -> str:
"""
调用 DeepSeek API 发送聊天请求
:param messages: 消息列表，格式为 [{"role": "system/user", "content": "..."}]
:return: LLM 返回的文本内容
"""
# 构造请求体，temperature 设为 0.1 保证输出稳定
data = json.dumps({
"model": MODEL,
"messages": messages,
"temperature": 0.1
}).encode("utf-8")
# 使用 urllib 发送 POST 请求（无需安装 requests 库）
req = urllib.request.Request(
API_URL,
data=data,
headers={
"Authorization": f"Bearer {API_KEY}",
"Content-Type": "application/json"
}
)
with urllib.request.urlopen(req) as resp:
result = json.loads(resp.read().decode("utf-8"))
# 提取 LLM 返回的文本内容
return result["choices"][0]["message"]["content"]
# ========================= 节点函数定义 =========================
def detect_node(state: ContentState) -> dict:
"""
节点1 - 内容检测节点
职责：调用 LLM 对用户输入进行敏感词检测
返回：敏感词列表、检测原因、风险评级
"""
print("【节点1】正在检测用户输入...")
user_input = state["user_input"]
# 构造提示词：要求 LLM 以 JSON 格式返回检测结果
messages = [
{"role": "system", "content": """你是一个内容安全检测专家。请对用户输入的内容进行敏感词检测，检测类别包括：脏话。
请严格以JSON格式返回，不要包含任何其他文字：
{
"sensitive_words": ["检测到的敏感词列表，没有则为空数组"],
"reason": "检测分析原因的详细说明",
"risk_level": "风险评级：低/中/高"
}"""},
{"role": "user", "content": f"请检测以下内容的敏感性：\n{user_input}"}
]
# 调用 LLM 并打印原始返回
result = call_deepseek(messages)
print(f" LLM原始返回: {result}")
# 解析 JSON，若 LLM 返回格式异常则使用默认值
try:
parsed = json.loads(result)
except json.JSONDecodeError:
parsed = {"sensitive_words": [], "reason": "LLM返回格式异常，默认通过", "risk_level": "低"}
# 返回状态更新：检测原始结果、敏感词、原因、风险评级
return {
"detection_result": result,
"sensitive_words": ", ".join(parsed.get("sensitive_words", [])) or "无",
"reason": parsed.get("reason", ""),
"risk_level": parsed.get("risk_level", "低")
}
def analyze_node(state: ContentState) -> dict:
"""
节点2 - 分析决策节点
职责：根据检测结果，由 LLM 判断是否需要人工审核
返回：是否需要审核、决策结果、决策原因
"""
print("\n【节点2】正在分析检测结果...")
# 构造提示词：要求 LLM 根据检测结果做出审核决策
messages = [
{"role": "system", "content": """你是一个内容审核决策专家。根据检测结果判断该内容是否需要人工审核。
判断标准：
- 包含政治敏感词汇 → 需要人工审核
- 包含脏话/侮辱性语言 → 需要人工审核
- 风险评级为"高" → 需要人工审核
- 风险评级为"中"且存在敏感词 → 需要人工审核
- 其他情况 → 通过
请严格以JSON格式返回，不要包含任何其他文字：
{
"need_review": true或false,
"decision": "需要人工审核" 或 "通过",
"analysis_reason": "决策原因的简要说明"
}"""},
# 将用户原始输入和检测节点的结果一起传给 LLM
{"role": "user", "content": f"""用户原始输入：{state['user_input']}
检测结果：
- 敏感词：{state['sensitive_words']}
- 原因：{state['reason']}
- 风险评级：{state['risk_level']}"""}
]
result = call_deepseek(messages)
print(f" LLM原始返回: {result}")
# 解析 JSON，异常时默认需要人工审核（安全优先）
try:
parsed = json.loads(result)
except json.JSONDecodeError:
parsed = {"need_review": True, "decision": "需要人工审核", "analysis_reason": "LLM返回格式异常，默认需审核"}
# 返回状态更新：是否需要审核、决策结果、追加决策原因
return {
"need_review": parsed.get("need_review", True),
"review_decision": parsed.get("decision", "需要人工审核"),
"reason": state["reason"] + " | 审核决策原因: " + parsed.get("analysis_reason", "")
}
def human_review_node(state: ContentState) -> dict:
"""
节点3a - 人工审核节点
职责：格式化输出需要人工审核的提醒信息
"""
# 拼接格式化的审核提醒
result = (
f"\n{'='*50}\n"
f" ⚠️ 需要人工审核\n"
f"{'='*50}\n"
f" 用户输入: {state['user_input']}\n"
f" 敏感词: {state['sensitive_words']}\n"
f" 风险评级: {state['risk_level']}\n"
f" 检测原因: {state['reason']}\n"
f" 审核建议: {state['review_decision']}\n"
f"{'='*50}"
)
print(result)
return {"final_result": result}
def pass_node(state: ContentState) -> dict:
"""
节点3b - 通过节点
职责：格式化输出内容通过检测的信息
"""
# 拼接格式化的通过信息
result = (
f"\n{'='*50}\n"
f" ✅ 内容通过检测\n"
f"{'='*50}\n"
f" 用户输入: {state['user_input']}\n"
f" 风险评级: {state['risk_level']}\n"
f" 检测结果: {state['review_decision']}\n"
f"{'='*50}"
)
print(result)
return {"final_result": result}
# ========================= 路由函数 =========================
def route_by_review(state: ContentState) -> str:
"""
条件路由：根据 analyze 节点的 need_review 字段决定下一步走向
:return: "human_review" 走人工审核节点，"pass" 走通过节点
"""
return "human_review" if state.get("need_review") else "pass"
# ========================= 构建 LangGraph 工作流 =========================
# 1. 创建状态图，绑定 ContentState 作为状态结构
graph = StateGraph(ContentState)
# 2. 注册四个节点
graph.add_node("detect", detect_node) # 内容检测节点
graph.add_node("analyze", analyze_node) # 分析决策节点
graph.add_node("human_review", human_review_node) # 人工审核节点
graph.add_node("pass", pass_node) # 通过节点
# 3. 添加边，定义流程走向
graph.add_edge(START, "detect") # 开始 → 检测
graph.add_edge("detect", "analyze") # 检测 → 分析
# 4. 条件分支：分析之后根据 need_review 动态路由
graph.add_conditional_edges(
"analyze", # 源节点
route_by_review, # 路由函数
{
"human_review": "human_review", # need_review=True → 人工审核
"pass": "pass", # need_review=False → 通过
}
)
# 5. 终止边：人工审核和通过节点都连接到 END
graph.add_edge("human_review", END)
graph.add_edge("pass", END)
# 6. 编译图，生成可执行的工作流
app = graph.compile()
# ========================= 运行工作流 =========================
# 测试输入（包含脏话和政治敏感词，预期触发人工审核）
user_input_2 = "这是一个美好的国家"
result_2 = app.invoke({
"user_input": user_input_2,
"detection_result": "",
"sensitive_words": "",
"reason": "",
"risk_level": "",
"need_review": False,
"review_decision": "",
"final_result": ""
})
# ========================= 生成并导出流程图 =========================
# 使用 Mermaid 语法生成可视化流程图，封装为 HTML 文件
mermaid_str = app.get_graph().draw_mermaid()
html_content = f"""<!DOCTYPE html>
<html>
<head>
<meta charset="utf-8">
<title>Content Detection Agent Flow</title>
</head>
<body>
<pre class="mermaid">{mermaid_str}</pre>
<script src="https://cdn.jsdelivr.net/npm/mermaid/dist/mermaid.min.js"></script>
<script>mermaid.initialize({{startOnLoad:true}});</script>
</body>
</html>"""
# 保存 HTML 流程图文件
output_path = r"e:\langGraphTest\001demo\flowchart_agent.html"
with open(output_path, "w", encoding="utf-8") as f:
f.write(html_content)
print(f"\n流程图已保存到: {output_path}")
print("请在浏览器中打开该文件查看流程图")
2、代码运行结果测试
【节点2】正在分析检测结果...
LLM原始返回: {
"need_review": false,
"decision": "通过",
"analysis_reason": "输入内容为中性描述，未包含政治敏感词或脏话，风险评级为低，无需人工审核。"
}
==================================================
✅ 内容通过检测
==================================================
用户输入: 这是一个美好的国家
风险评级: 低
检测结果: 通过
==================================================
流程图已保存到: e:\langGraphTest\001demo\flowchart_agent.html
请在浏览器中打开该文件查看流程图
3、代码执行的流程图
有“AI”的1024 = 2048，欢迎大家加入2048 AI社区
更多推荐
6
0- 0
已为社区贡献4条内容
所有评论(0)