LangGraph同步与异步流实现解析

1. LangGraph 同步与异步流实现解析

作为一名长期从事AI应用开发的工程师，我最近在构建对话系统时遇到了一个典型问题：如何处理包含多个步骤的复杂用户请求？比如用户问"北京今天气温怎么样要穿什么？"，这需要先识别城市、查询天气、再给出穿衣建议。传统方式要么写一堆if-else，要么用回调地狱，维护起来非常痛苦。直到我发现了LangGraph这个神器，它让我能用图的方式优雅地组织业务流程，还能自由选择同步或异步执行模式。

2. 项目架构与核心设计

2.1 系统功能链路设计

这个天气穿衣建议系统的核心流程非常清晰：

1. 用户输入处理：接收自然语言查询
2. 意图识别：判断是否包含城市和天气相关关键词
3. 天气查询：调用模拟天气API获取温度
4. 建议生成：基于温度给出穿衣建议
5. 结果返回：通过Flask API返回JSON响应

这种链式处理非常适合用LangGraph的StateGraph来实现。StateGraph的核心思想是将业务流程建模为状态转换图，每个节点处理部分逻辑并更新共享状态。

2.2 状态(State)设计

State是整个流程的共享内存，我设计了一个WeatherState类型：

python复制class WeatherState(TypedDict):
    user_input: str  # 原始用户输入
    city: Optional[str]  # 识别的城市
    temperature: Optional[int]  # 查询到的温度 
    advice: Optional[str]  # 生成的穿衣建议

这种设计的好处是：

• 类型安全：使用TypedDict确保状态字段明确
• 可扩展：随时可以添加新字段而不破坏现有逻辑
• 可追溯：完整记录处理过程中的中间状态

3. 同步实现详解

3.1 业务逻辑节点实现

同步版本包含三个核心节点：

3.1.1 城市提取节点

python复制def extract_city_and_weather(state: WeatherState):
    text = state["user_input"]
    cities = ["北京", "上海", "广州", "深圳", "杭州"]  # 简化版城市词典
    
    found_city = next((c for c in cities if c in text), None)
    has_weather_intent = any(k in text for k in ["天气", "气温", "热", "冷", "穿什么"])
    
    return {"city": found_city} if found_city and has_weather_intent else {"city": None}

实际项目中建议接入专业的NLP实体识别服务，这里用简单实现演示原理

3.1.2 温度查询节点

python复制def fetch_temperature(state: WeatherState):
    if not state["city"]:
        return {"temperature": None}
    
    # 模拟天气API返回
    fake_weather_db = {
        "北京": 5, "上海": 13,