上一篇 从 LangChain 的局限，到 Agent Server 的进化

一、项目背景

在传统的快递系统中，配送流程通常是固定的，但在实际场景中：

不同目的地 → 分拣路径不同 不同优先级 → 运输方式不同（陆运 / 空运） 状态需要持续更新 → 且保留完整历史

那么问题来了：

如何用一种“流程化 + 可扩展”的方式来建模这个系统？

这正是 LangGraph 擅长的场景。

---

二、核心思路

我们用 LangGraph 构建一个“状态驱动的流程图”，核心拆成三部分：

1. State（状态）

用于描述包裹当前信息

2. Node（节点）

表示每个物流环节（揽收 / 分拣 / 配送）

3. Edge（边）

控制流程流转（包括条件分支）

---

三、状态设计（State）

# 1.定义状态State
class PackageState(TypedDict):
#包裹id
    package_id:str
#始发站
    origin:str
#目的地
    destination:str
#配送状态
    status:str #"待揽收", "运输中", "派送中", "已签收
#流转历史
    history:Annotated[List[str],add] 
#总里程
    total_distance:Annotated[int,add]
#配送详情  加急、标准配送
    priority:str
    

---

🔥 重点：
Annotated + add history:

Annotated[List[str], add]

total_distance: Annotated[int, add]

解释：LangGraph 专用指令，add = 追加模式
状态更新时不是覆盖，而是追加

比如：

history 会不断记录轨迹
total_distance 会自动累加

这就是 LangGraph 的“状态增强能力”

---

四、节点设计（Node）

1. "揽收站"节点

def receive_package(state: PackageState):

    return {
        "status": "已揽收",
        "history": [f"在{state['origin']}处揽收"]
    }

更新状态 + 写入第一条轨迹

---

2. “分拣中心”

def sort_package(state: PackageState):
    destination = state["destination"]

    if "北京" in destination:
        next = "北京分拣中心"
    elif "广东" in destination:
        next = "上海分拣中心"
    else:
        next = "其他地区分拣中心"

    return {
        "status": "已分拣",
        "history": [f"将分拣到{next}"]
    }

根据目的地动态决定分拣路径

---

3. ”配送“节点（关键分支）

标准配送（陆运）

def standard_delivery(state: PackageState):
    return {
        "status": "运输中",
        "history": ["标准陆运"],
        "total_distance": 500
    }

---

加急配送（空运）

def express_delivery(state: PackageState):
    return {
        "status": "加急运输中",
        "history": ["加急空运"],
        "total_distance": 900
    }

---

4. ”派送“完成

def final_delivery(state: PackageState):
    return {
        "status": "已签收",
        "history": [f"已送达至{state['destination']}"]
    }

五、构建流程图（Graph）

1. 初始化

delivery = StateGraph(PackageState)

---

2. 添加节点

delivery.add_node("揽收站", receive_package)
delivery.add_node("分拣中心", sort_package)
delivery.add_node("派送站", final_delivery)
delivery.add_node("标准配送", standard_delivery)
delivery.add_node("加急配送", express_delivery)

---

3. 定义流程

delivery.add_edge(START, "揽收站")
delivery.add_edge("揽收站", "分拣中心")

---

1. 条件路由

def select_delivery(state: PackageState):
    if state["priority"] == "加急":
        return "加急配送"
    else:
        return "标准配送"
delivery.add_conditional_edges(
    "分拣中心",
    select_delivery,
    ["加急配送", "标准配送"]
)

这一步实现了：

根据状态动态决定流程走向是 "加急"还是"标准"配送

---

4. 完整流程闭环

delivery.add_edge("标准配送", "派送站")
delivery.add_edge("加急配送", "派送站")
delivery.add_edge("派送站", END)

---

5.编译图

delivery_compile=delivery.compile()

.compile() 是 LangGraph 内部的 StateGraph 类提供的方法

可以把流程图编译成可运行的程序

---

六、运行结果分析

测试数据

	test_packages = [
	    {
	        "package_id": "A001",
	        "origin": "北京",
	        "destination": "嘉兴",
	        "priority": "普通",
	        "history": [],
	        "total_distance": 0
	    },
	    {
	        "package_id": "C002",
	        "origin": "广州",
	        "destination": "乌鲁木齐",
	        "priority": "加急",
	        "history": [],
	        "total_distance": 0
	    }
	]

---

运行结果

P001（普通件）
最终状态: 已签收
配送历史: [
	  '在北京处揽收',
	  '将分拣到上海分拣中心',
	  '标准陆运',
	  '已送达至上海'
]
总里程: 500

---

标准配送总流程 ：

揽收 → 分拣 → 标准配送 → 派送完成

P002（加急件）
最终状态: 已签收
配送历史: [
	  '在广州处揽收',
	  '将分拣到其他地区分拣中心',
	  '加急空运',
	  '已送达至乌鲁木齐'
]
总里程: 900

加急配送总流程：

揽收 → 分拣 → 加急配送 → 派送完成

---

今天的分享就到这了，下期再见~