-python-LangGraph框架（3-33-LangGraph多分枝实战）

剪影AS

234人浏览 · 2026-07-03 17:04:35

剪影AS · 2026-07-03 17:04:35 发布

、先明确核心场景：为什么需要多分支？

我们以“户外活动适宜性评估”为具体场景：判断一个城市某一时刻是否适合户外活动，需要两个核心数据支撑——天气数据（温度、天气状况）和时间数据（当前时段、时段类型），这两个数据的获取的是相互独立的，无需等待其中一个完成再执行另一个，且最终的决策必须结合两个数据才能得出，缺一不可。

这种“多任务并行获取、单节点汇聚决策”的场景，正是LangGraph多分支的优势所在。它不像线性工作流那样只能串行执行，而是可以让多个专业化任务同时推进，既提升了执行效率，又能实现“分工明确、协同决策”的逻辑，这也是LangGraph超越传统工作流引擎的核心亮点之一。

二、LangGraph多分支核心特点：分工、并行、汇聚

LangGraph的核心哲学是“将程序逻辑从过程指令转化为状态拓扑空间”，而多分支作为其核心能力，主要体现在三个方面，结合我们的实战案例就能轻松理解：

1. 专业化分工：每个分支只做一件事

多分支的核心前提是“职责拆分”，避免一个节点承担过多任务，提升代码的可维护性和复用性。在我们的案例中，我们将整个任务拆分为两个专业化分支，每个分支专注于自己的核心职责，不越界、不冗余：

- 分支一：专注获取天气相关数据，仅输出原始天气信息，不做任何决策判断，确保数据的纯粹性；

- 分支二：专注获取时间相关数据，仅输出原始时间信息，同样不参与决策，只负责数据采集。

这种分工模式，不仅让每个分支的逻辑更简洁，也便于后续的扩展——比如后续需要新增“空气质量”分支，只需新增一个专业化节点即可，无需修改原有逻辑，体现了LangGraph模块化扩展的优势。

2. 并行执行：提升效率，无需相互等待

这是LangGraph多分支最实用的特点之一。两个专业化分支的任务是相互独立的，没有依赖关系，因此LangGraph会自动触发两个分支并行执行，而不是串行等待一个分支完成再执行另一个。

举个直观的例子：如果获取天气数据需要1秒，获取时间数据需要1秒，线性执行总共需要2秒，而并行执行只需1秒即可完成两个数据的获取，大幅提升了整体任务的执行效率。LangGraph的异步执行引擎会自动管理并行逻辑，包括同时启动多个节点、等待所有分支返回结果，还能通过内置信号量控制并发度，避免资源过载。

3. 汇聚决策：多分支结果融合，输出最终答案

多分支并行执行后，会产生多个原始数据，这些数据单独存在时没有实际意义——比如只知道天气晴朗，不知道具体时段，无法判断是否适合户外活动；只知道时段合适，不知道天气状况，也无法做出准确决策。

因此，LangGraph多分支的核心闭环是“汇聚节点”：它会收集所有分支的输出结果，对数据进行整理、校验，再结合预设的决策逻辑，融合所有分支的数据，输出最终的完整答案。这个汇聚节点是整个多分支流程的“大脑”，负责整合所有专业化分支的成果，实现“1+1>2”的协同效果。

三、LangGraph多分支实现方式：4步搭建完整流程

结合我们的实战案例，LangGraph多分支的实现逻辑非常清晰，无需复杂的配置，核心只需4步，就能搭建起“分工-并行-汇聚”的完整工作流，贴合LangGraph“状态驱动”的核心编程模型：

第一步：定义核心节点，明确分支职责

首先需要定义整个图结构的核心节点，每个节点对应一个具体的任务，节点的职责必须清晰、单一。我们的案例中，共定义了4个核心节点，各司其职：

- 入口节点：作为整个流程的启动器，负责接收用户的查询请求，同时触发两个专业化分支的并行执行，相当于多分支的“总开关”；

- 两个专业化分支节点：分别负责获取天气、时间数据，输出原始数据，不参与决策；

- 汇聚决策节点：负责收集两个分支的原始数据，校验数据的完整性，再结合决策逻辑，输出最终的户外活动适宜性评估报告。

节点的定义遵循“单一职责原则”，这也是LangGraph多分支能够灵活扩展的基础——每个节点都是独立的模块，可单独修改、替换，不影响其他节点的运行。

第二步：搭建图结构，连接分支与汇聚节点

LangGraph的核心是“图结构”，节点定义完成后，需要通过“边”将这些节点连接起来，明确流程的执行顺序和关联关系，这也是实现多分支的关键：

- 入口节点与两个分支节点建立连接：确保入口节点启动后，能同时触发两个分支并行执行；

- 两个分支节点与汇聚决策节点建立连接：确保两个分支的输出结果，能准确传递到汇聚节点，供后续决策使用；

- 汇聚决策节点与结束节点连接：确保决策完成后，整个流程正常终止。

这种连接方式，形成了“入口→多分支并行→汇聚→结束”的完整闭环，图结构的可视化也让整个流程的逻辑更加清晰，便于后期排查问题和优化。

第三步：配置分支逻辑，确保数据规范

每个专业化分支节点，都需要配置对应的处理逻辑，确保输出的数据格式统一、规范，便于汇聚节点进行处理。比如：

对于天气分支，我们会规范其输出格式，确保每次返回的数据都包含温度、天气状况、城市等核心信息；对于时间分支，同样规范输出格式，包含小时、分钟、时段类型、城市等信息。

同时，我们还会对分支的输出结果进行简单的清理和标识，避免不同分支的输出数据混淆，确保汇聚节点能准确识别、提取每个分支的数据——这一步看似简单，却能有效避免后续数据提取时出现错误，提升整个流程的稳定性。

第四步：实现汇聚逻辑，完成多分支协同决策

汇聚节点是多分支流程的核心，其核心逻辑分为两步：

1. 数据收集与校验：汇聚节点会自动收集两个分支的输出结果，对数据进行校验，判断是否存在缺失——如果某个分支的数据缺失，会提示决策失败，确保最终决策的准确性；

2. 多数据融合决策：在确保数据完整的前提下，汇聚节点会结合预设的决策逻辑，同时考虑天气和时间两个因素，进行综合判断，最终输出清晰、易懂的评估报告，包括决策依据、综合结论和具体建议。

这里需要注意的是，汇聚节点的决策逻辑必须依赖所有分支的数据，单一分支的数据无法完成决策——这也体现了多分支协作的意义，通过多个专业化数据的融合，让决策更加全面、合理。

四、实战总结：LangGraph多分支的优势与适用场景

通过这个户外活动评估的实战案例，我们能清晰感受到LangGraph多分支的优势，它不仅解决了传统线性工作流效率低、扩展性差的问题，还带来了更灵活、更易维护的开发体验，其核心优势可总结为三点：

1. 效率提升：多分支并行执行，大幅缩短整体任务的执行时间，尤其适合多任务独立的场景；

2. 扩展性强：新增分支只需新增节点并建立连接，无需修改原有逻辑，适配复杂场景的迭代需求，这也是LangGraph被广泛应用于企业级场景的重要原因之一；

3. 逻辑清晰：图结构的可视化的，让多分支的执行流程、节点关联一目了然，便于开发、调试和后期维护，同时状态驱动的模型也让流程控制更加灵活。

而LangGraph多分支的适用场景，远不止我们案例中的户外活动评估，只要是需要“多任务并行、多数据融合决策”的场景，都可以使用它来实现，比如：

- 客服自动化：将用户请求拆分为“意图识别”“信息查询”“回复生成”多个分支，并行处理，提升响应速度；

- 多维度数据分析：多个分支分别获取不同维度的数据，汇聚节点进行融合分析，输出综合报告；

- 智能代理开发：多个专业化代理并行工作，汇聚节点整合各代理的结果，完成复杂任务。

五、最后想说的话

LangGraph的多分支能力，本质上是将复杂的任务拆解为多个可并行的专业化子任务，再通过汇聚节点实现协同决策，它让大模型应用的开发从“线性思维”转向“图结构思维”，不仅提升了开发效率，也让复杂工作流的设计变得更加直观、灵活。

本文的案例虽然简单，但涵盖了LangGraph多分支的核心实现逻辑——节点定义、图结构搭建、分支配置、汇聚决策，只要掌握了这四个核心步骤，就能轻松应对大多数多分支协作的场景。

如果是刚接触LangGraph，建议从简单的多分支场景入手，熟悉节点、边、状态的核心概念，再逐步尝试更复杂的循环、条件分支等功能，相信你会发现LangGraph在复杂大模型应用开发中的强大之处。

后续我也会分享更多LangGraph的实战技巧，比如条件分支、循环逻辑的实现，感兴趣的可以关注一下～

代码实现：

100

101

102

103

104

105

106

107

108

109

110

111

112

113

114

115

116

117

118

119

120

121

122

123

124

125

126

127

128

129

130

131

132

133

134

135

136

137

138

139

140

141

142

143

144

145

146

147

148

149

150

151

152

153

154

155

156

157

158

159

160

161

162

163

164

165

166

167

168

169

170

171

172

173

174

175

176

177

178

179

180

181

182

183

184

185

186

187

188

189

190

191

192

193

194

195

196

197

198

199

200

201

202

203

204

205

206

207

208

209

210

211

212

213

214

215

216

217

218

219

220

221

222

223

224

225

226

227

228

229

230

231

232

233

234

235

236

237

238

239

240

241

242

243

244

245

246

247

248

249

250

251

252

253

254

255

256

257

258

259

260

261

262

263

264

265

266

267

268

269

270

271

272

273

274

275

276

277

278

279

280

281

282

283

284

285

286

287

288

289

290

291

292

293

294

295

296

297

298

299

300

301

302

303

304

305

306

307

308

309

310

311

312

313

314

315

316

317

318

319

320

321

322

323

324

325

326

327

328

329

330

331

332

333

import json

import re

from typing import List, Dict, Any, Optional

from langchain_core.messages import HumanMessage, AIMessage, SystemMessage

from langchain_openai import ChatOpenAI

from langgraph.graph import MessageGraph, END

def remove_think_tags(text):

"""

移除字符串中<think></think>标签及其内容

Args:

text: 包含think标签的字符串

Returns:

移除think标签及其内容后的字符串

"""

# 支持多行内容和标签可能有的属性

pattern = r'<think[^>]*>.*?</think>'

result = re.sub(pattern, '', text, flags=re.DOTALL)

# 清理空白字符

result = result.strip()

result = re.sub(r'\n{3,}', '\n\n', result) # 限制连续换行

return result

def extract_json_from_llm_response(text: str) -> Optional[Dict[str, Any]]:

"""

从LLM响应中健壮地提取JSON对象

处理多种格式：

1. 纯JSON: {"key": "value"}

2. Markdown代码块: ```json {...} ```

3. 带前缀/后缀的JSON: "Result: {...}"

"""

text = remove_think_tags(text.strip())

# 尝试1：直接解析整个字符串

try:

return json.loads(text)

except json.JSONDecodeError:

pass

# 尝试2：提取Markdown代码块中的JSON

code_block_patterns = [

r'```(?:json)?\s*({.*?})\s*```', # ```json {...} ```

r'```\s*({.*?})\s*```', # ``` {...} ```

r'({.*})' # 直接匹配花括号内容（最后手段）

]

for pattern in code_block_patterns:

match = re.search(pattern, text, re.DOTALL)

if match:

try:

return json.loads(match.group(1))

except json.JSONDecodeError:

continue

# 尝试3：查找第一个{和最后一个}之间的内容（处理嵌套）

start = text.find('{')

end = text.rfind('}')

if start != -1 and end != -1 and start < end:

candidate = text[start:end + 1]

try:

return json.loads(candidate)

except json.JSONDecodeError:

pass

print(f"⚠️ JSON提取失败，原始响应: {text[:100]}...")

return None

# ==================== 1. LLM 配置 ====================

DEEPSEEK_API_KEY = "123" # 替换为实际的 API Key

llm = ChatOpenAI(

api_key=DEEPSEEK_API_KEY,

base_url="http://192.168.0.100:8087/v1",

model="qwen3.5-35b-gptq",

temperature=0.3,

max_tokens=20000,

)

# ==================== 2. 专业化分支节点（只提供原始数据） ====================

def source_node(state: List[Any]) -> List[Any]:

"""

源节点：触发两个专业化分支

返回原始状态，LangGraph 会自动将消息分发到所有出边（实现并行）

"""

print("\n[source_node] 触发天气和时间两个专业化分支")

return state

def weather_branch(state: List[Any]) -> AIMessage:

"""

天气分支：仅返回原始天气数据（无决策能力）

输出示例: {"temperature": 22, "condition": "sunny", "city": "北京"}

"""

print("\n[weather_branch] 专注提取天气数据（无决策能力）")

last_msg = state[-1].content if state else ""

system_prompt = (

"你是一个专业天气数据提取器。严格遵守：\n"

"1. 只提取天气相关数据，不做任何决策或建议\n"

"2. 输出纯JSON，不要任何额外文本或Markdown代码块\n"

"3. JSON格式: {\"temperature\": 整数, \"condition\": \"sunny/cloudy/rainy/snowy\", \"city\": \"城市名\"}"

)

user_prompt = f"用户问题：{last_msg}\n\n请返回该城市的模拟天气数据："

response = llm.invoke([

SystemMessage(content=system_prompt),

HumanMessage(content=user_prompt)

])

# 清理响应并添加分支标识

cleaned = remove_think_tags(response.content.strip())

return AIMessage(content=f"[WEATHER]{cleaned}")

def time_branch(state: List[Any]) -> AIMessage:

"""

时间分支：仅返回原始时间数据（无决策能力）

输出示例: {"hour": 15, "minute": 30, "period": "afternoon", "city": "北京"}

"""

print("\n[time_branch] 专注提取时间数据（无决策能力）")

last_msg = state[-1].content if state else ""

system_prompt = (

"你是一个专业时间数据提取器。严格遵守：\n"

"1. 只提取当前时间数据，不做任何决策或建议\n"

"2. 输出纯JSON，不要任何额外文本或Markdown代码块\n"

"3. JSON格式: {\"hour\": 0-23, \"minute\": 0-59, \"period\": \"morning/afternoon/evening/night\", \"city\": \"城市名\"}"

)

user_prompt = f"用户问题：{last_msg}\n\n请返回该城市的模拟当前时间："

response = llm.invoke([

SystemMessage(content=system_prompt),

HumanMessage(content=user_prompt)

])

cleaned = remove_think_tags(response.content.strip())

return AIMessage(content=f"[TIME]{cleaned}")

def decision_sink(state: List[Any]) -> AIMessage:

"""

汇聚决策节点：必须结合天气+时间数据才能做出完整回答

单一分支数据无法回答"是否适合户外活动"这类综合问题

"""

print("\n[decision_sink] 综合天气+时间数据进行智能决策")

# 从state中提取带标识的分支结果（兼容并行执行的顺序不确定性）

weather_data = None

time_data = None

# 从最新消息向前查找（确保获取最新分支输出）

for msg in reversed(state):

if isinstance(msg, AIMessage):

content = msg.content

if "[WEATHER]" in content and weather_data is None:

json_str = content.replace("[WEATHER]", "").strip()

weather_data = extract_json_from_llm_response(json_str)

print(f" ✓ 天气数据提取: {weather_data}")

elif "[TIME]" in content and time_data is None:

json_str = content.replace("[TIME]", "").strip()

time_data = extract_json_from_llm_response(json_str)

print(f" ✓ 时间数据提取: {time_data}")

# 验证必要数据

if not weather_data or not time_data:

missing = []

if not weather_data: missing.append("天气数据")

if not time_data: missing.append("时间数据")

return AIMessage(content=f"❌ 决策失败：缺少{'和'.join(missing)}，无法进行综合判断")

# === 核心：必须结合两个分支数据才能做出合理决策 ===

decision = make_outdoor_activity_decision(weather_data, time_data)

return AIMessage(content=decision)

def make_outdoor_activity_decision(weather: Dict[str, Any], time: Dict[str, Any]) -> str:

"""

基于天气+时间的综合决策（必须两个数据源）

单独任一数据都无法安全判断户外活动适宜性

"""

city = weather.get("city", time.get("city", "未知城市"))

temp = weather.get("temperature", "N/A")

condition = weather.get("condition", "unknown")

hour = time.get("hour", -1)

period = time.get("period", "unknown")

# 决策逻辑：必须同时考虑天气和时间

factors = []

recommendations = []

suitability_score = 0

# ===== 时间因素分析（单独无法决策）=====

if hour < 6 or hour > 21:

factors.append(f"⚠️ {period}时段({hour}点)：光线不足/温度较低")

suitability_score -= 30

elif 6 <= hour <= 9:

factors.append(f"✅ 早晨{hour}点：适合晨练，空气清新")

suitability_score += 20

elif 15 <= hour <= 17:

factors.append(f"✅ 下午{hour}点：阳光适宜，温度舒适")

suitability_score += 25

else:

factors.append(f"⚪ {period}时段({hour}点)：时间中性")

# ===== 天气因素分析（单独无法决策）=====

if condition == "rainy":

factors.append("⚠️ 正在下雨：路面湿滑，易感冒")

suitability_score -= 40

elif condition == "snowy":

factors.append("⚠️ 正在下雪：路面积雪，能见度低")

suitability_score -= 45

elif condition == "sunny":

if isinstance(temp, int) and temp > 32:

factors.append(f"⚠️ 晴天但高温({temp}°C)：中暑风险高")

suitability_score -= 25

elif isinstance(temp, int) and 20 <= temp <= 28:

factors.append(f"✅ 晴朗舒适({temp}°C)：紫外线适中")

suitability_score += 30

else:

factors.append(f"⚪ 晴天({temp}°C)：天气良好")

suitability_score += 15

elif condition == "cloudy":

factors.append(f"✅ 多云({temp}°C)：紫外线弱，体感舒适")

suitability_score += 25

# ===== 综合决策（必须两个数据）=====

if suitability_score >= 30:

conclusion = "🟢 非常适合户外活动！"

recommendations = [

"建议进行：慢跑、骑行、公园散步",

"注意：及时补充水分"

]

elif suitability_score >= 10:

conclusion = "🟡 适合户外活动，需注意条件"

recommendations = [

"建议进行：短途散步、户外休闲",

"注意：根据天气准备相应装备（雨具/防晒）"

]

elif suitability_score >= -10:

conclusion = "🟠 谨慎进行户外活动"

recommendations = [

"建议：缩短户外时间，选择遮蔽区域",

"避免：剧烈运动、长时间暴露"

]

else:

conclusion = "🔴 不建议户外活动"

recommendations = [

"建议：室内活动替代",

"原因：天气/时间条件不适宜"

]

# 生成综合报告

report = (

f"📍 {city}户外活动适宜性综合评估\n"

f"{'─' * 40}\n"

f"🌤️ 天气数据: {temp}°C, {condition}\n"

f"⏰ 时间数据: {hour}点 ({period})\n"

f"{'─' * 40}\n"

f"💡 决策依据:\n"

)

for factor in factors:

report += f" • {factor}\n"

report += f"{'─' * 40}\n"

report += f"🎯 综合结论: {conclusion}\n"

report += f"📋 建议:\n"

for rec in recommendations:

report += f" • {rec}\n"

return report

# ==================== 3. 构建专业化多分支图 ====================

graph = MessageGraph()

graph.add_node("source", source_node)

graph.add_node("weather_branch", weather_branch)

graph.add_node("time_branch", time_branch)

graph.add_node("decision_sink", decision_sink)

graph.set_entry_point("source")

graph.add_edge("source", "weather_branch")

graph.add_edge("source", "time_branch")

graph.add_edge("weather_branch", "decision_sink")

graph.add_edge("time_branch", "decision_sink")

graph.add_edge("decision_sink", END)

app = graph.compile()

#画图

print(app.get_graph().draw_ascii())

# ==================== 4. 执行测试 ====================

if __name__ == "__main__":

print("=" * 70)

print("✅ 真实多分支协作测试：需要天气+时间数据共同决策")

print("=" * 70)

test_cases = [

# "北京现在适合户外跑步吗？",

# "上海下午四点去公园散步合适吗？",

# "广州晚上九点适合夜跑吗？",

"成都中午十二点适合户外野餐吗？"

]

for i, query in enumerate(test_cases, 1):

print(f"\n{'=' * 70}")

print(f"_TestCase {i}: {query}_")

print('=' * 70)

inputs = [HumanMessage(content=query)]

for event in app.stream(inputs):

node = list(event.keys())[0]

msg = event[node]

# 打印中间节点输出（简化显示）

if node in ["weather_branch", "time_branch"]:

content_preview = msg.content.replace("[WEATHER]", "").replace("[TIME]", "").strip()

prefix = "🌤️" if node == "weather_branch" else "⏰"

print(f"\n[→ {node}] {prefix} {content_preview[:80]}...")

elif node == "decision_sink":

print(f"\n[→ {node}] 🎯 最终决策报告:\n")

print(msg.content)

结果输出：

+-----------+
| __start__ |
+-----------+
*
*
*
+--------+
| source |
+--------+*
** **
** **
** **
+-------------+ +----------------+
| time_branch | | weather_branch |
+-------------+ +----------------+
** **
** **
** **
+---------------+
| decision_sink |
+---------------+
*
*
*
+---------+
| __end__ |
+---------+
======================================================================
✅ 真实多分支协作测试：需要天气+时间数据共同决策
======================================================================

======================================================================
_TestCase 1: 成都中午十二点适合户外野餐吗？_
======================================================================

AI Agent技术社区

Agent 垂直技术社区，欢迎活跃、内容共建。

更多推荐

SeedVR2终极指南：如何用单步扩散模型实现专业级视频修复

还在为老旧视频的模糊画质而烦恼吗？面对抖动严重的监控录像或褪色的家庭影像，你是否曾希望有一种简单高效的修复方案？SeedVR2正是为解决这些痛点而生的一站式视频修复工具。作为字节跳动推出的先进视频修复模型，它能够在单步推理中实现专业级的视频质量提升，彻底改变了传统视频修复的工作流程。## 核心价值：为什么选择SeedVR2进行视频修复？**传统视频修复的三大瓶颈**：1. **计算成本高

AI Agent技术社区

深度解析Amlogic S9xxx电视盒子Armbian系统：从闲置设备到Linux服务器的终极实战指南

你是否拥有一台闲置的Amlogic S9xxx系列电视盒子？想让它摆脱单一的视频播放功能，变身为功能强大的Linux服务器、开发平台或家庭媒体中心？本指南将为你详细解析如何通过Armbian系统实现这一转变，让你的电视盒子重获新生，成为高性能的Linux设备。Armbian系统基于Debian/Ubuntu构建，专为ARM芯片优化，提供了完整的Linux生态支持。## 🎯 核心价值与独特优势

AI Agent技术社区

CANN/runtime Event管理API

本章节描述 CANN Runtime 的 Event 管理接口，用于事件的创建、记录、同步、计时及 IPC 跨进程共享。- [`aclError aclrtCreateEvent(aclrtEvent *event)`](#aclrtCreateEvent)：创建Event，创建出来的Event可用于统计两个Event之间的耗时、多Stream之间的任务同步等场景。- [`aclError