在这里插入图片描述

该 Demo 的“系统功能链路”为：

  1. 用户发消息：“北京今天气温怎么样要穿什么？”
  2. StateGraph 判断是否提到“城市 + 气温”
  3. 若提到城市 → 调用“气温 API”（我们用模拟函数代替）
  4. **调用 RAG（伪）搜索该气温下穿什么”
  5. 返回最终回复给用户（Flask API 返回 JSON）

1.同步执行

代码

# ============================================================
# 1. 导入
# ============================================================
from flask import Flask, request, jsonify
from langgraph.graph import StateGraph
from typing import TypedDict, Optional

# ============================================================
# 2. 定义 State（系统的“全局记忆结构”）
# ============================================================
class WeatherState(TypedDict):
    user_input: str
    city: Optional[str]
    temperature: Optional[int]
    advice: Optional[str]


# ============================================================
# 3. ----------- 业务逻辑函数（可被 StateGraph 调用）-----------
# ============================================================

# ----- step1：NLU 识别是否出现城市和“气温”相关词 -----
def extract_city_and_weather(state: WeatherState):
    text = state["user_input"]

    # 简化的城市词表（真实项目可接入实体识别模型）
    cities = ["北京", "上海", "广州", "深圳", "杭州"]

    found_city = None
    for c in cities:
        if c in text:
            found_city = c
            break

    # 判断用户是否提到天气/气温
    has_weather_intent = any(k in text for k in ["天气", "气温", "热", "冷", "穿什么"])

    if found_city and has_weather_intent:
        return {"city": found_city}
    else:
        # 没城市或没天气意图 → 不进入后续步骤
        return {"city": None}


# ----- step2：获取城市气温（模拟 API ）-----
def fetch_temperature(state: WeatherState):
    city = state["city"]
    if not city:
        return {"temperature": None}

    # 假设模拟温度
    fake_weather_db = {
        "北京": 5,
        "上海": 13,
        "广州": 21,
        "深圳": 22,
        "杭州": 10,
    }
    return {"temperature": fake_weather_db.get(city, 15)}


# ----- step3：RAG 查询穿衣建议（我们用简单规则代替） -----
def rag_fetch_clothing_advice(state: WeatherState):
    temp = state["temperature"]

    if temp is None:
        return {"advice": "未识别到城市或天气相关内容。"}

    if temp <= 5:
        a = "天气很冷，建议穿厚羽绒服、围巾、手套。"
    elif temp <= 15:
        a = "有点冷，建议穿风衣、薄棉服、长袖。"
    elif temp <= 25:
        a = "天气适中，可以穿短袖或薄外套。"
    else:
        a = "天气较热，可以穿短袖短裤。"

    return {"advice": a}


# ============================================================
# 4. ----------- 构建 LangGraph StateGraph -------------------
# ============================================================
graph = StateGraph(WeatherState)

graph.add_node("extract_city", extract_city_and_weather)
graph.add_node("get_temp", fetch_temperature)
graph.add_node("rag_advice", rag_fetch_clothing_advice)

graph.set_entry_point("extract_city")

# 条件分支：
# 若识别到 city，则进入天气查询，否则直接返回 advice
def has_city(state: WeatherState):
    return "go" if state["city"] else "stop"

graph.add_conditional_edges(
    "extract_city",
    has_city,
    {
        "go": "get_temp",
        "stop": "rag_advice"
    }
)

# 获取温度之后一定进入 RAG
graph.add_edge("get_temp", "rag_advice")

# 编译图
app_graph = graph.compile()


# ============================================================
# 5. -------------------- Flask API ---------------------------
# ============================================================
app = Flask(__name__)

@app.route("/ask", methods=["POST"])
def ask_weather():
    data = request.json
    user_input = data.get("text", "")

    # 执行 LangGraph
    result = app_graph.invoke({
        "user_input": user_input,
        "city": None,
        "temperature": None,
        "advice": None
    })

    return jsonify({
        "city": result["city"],
        "temperature": result["temperature"],
        "advice": result["advice"]
    })


# ============================================================
# 6. 启动 Flask
# ============================================================
if __name__ == "__main__":
    app.run(host="0.0.0.0", port=5000)

使用

启动服务后发送：

curl -X POST http://127.0.0.1:5000/ask \
    -H "Content-Type: application/json" \
    -d '{"text":"北京今天冷吗穿什么？"}'

返回结果示例：

{
  "city": "北京",
  "temperature": 5,
  "advice": "天气很冷，建议穿厚羽绒服、围巾、手套。"
}

2.异步流执行

代码

# ============================================================
# 1. 导入
# ============================================================
from flask import Flask, request, jsonify, Response
from langgraph.graph import StateGraph
from typing import TypedDict, Optional
import asyncio
import json

# ============================================================
# 2. 定义 State
# ============================================================
class WeatherState(TypedDict):
    user_input: str
    city: Optional[str]
    temperature: Optional[int]
    advice: Optional[str]

# ============================================================
# 3. 业务逻辑节点
# ============================================================
def extract_city_and_weather(state: WeatherState):
    text = state["user_input"]
    cities = ["北京", "上海", "广州", "深圳", "杭州"]
    found_city = None
    for c in cities:
        if c in text:
            found_city = c
            break
    has_weather_intent = any(k in text for k in ["天气", "气温", "热", "冷", "穿什么"])
    if found_city and has_weather_intent:
        return {"city": found_city}
    else:
        return {"city": None}

async def fetch_temperature(state: WeatherState):
    await asyncio.sleep(0.2)  # 模拟延迟
    city = state["city"]
    if not city:
        return {"temperature": None}
    fake_weather_db = {
        "北京": 5,
        "上海": 13,
        "广州": 21,
        "深圳": 22,
        "杭州": 10,
    }
    return {"temperature": fake_weather_db.get(city, 15)}

async def rag_fetch_clothing_advice(state: WeatherState):
    await asyncio.sleep(0.2)  # 模拟 RAG 查询延迟
    temp = state["temperature"]
    if temp is None:
        return {"advice": "未识别到城市或天气相关内容。"}
    if temp <= 5:
        a = "天气很冷，建议穿厚羽绒服、围巾、手套。"
    elif temp <= 15:
        a = "有点冷，建议穿风衣、薄棉服、长袖。"
    elif temp <= 25:
        a = "天气适中，可以穿短袖或薄外套。"
    else:
        a = "天气较热，可以穿短袖短裤。"
    return {"advice": a}

# ============================================================
# 4. 构建 StateGraph
# ============================================================
graph = StateGraph(WeatherState)
graph.add_node("extract_city", extract_city_and_weather)
graph.add_node("get_temp", fetch_temperature)
graph.add_node("rag_advice", rag_fetch_clothing_advice)
graph.set_entry_point("extract_city")

# 条件分支
def has_city(state: WeatherState):
    return "go" if state["city"] else "stop"

graph.add_conditional_edges(
    "extract_city",
    has_city,
    {"go": "get_temp", "stop": "rag_advice"}
)
graph.add_edge("get_temp", "rag_advice")

# 编译图
app_graph = graph.compile()

# ============================================================
# 5. Flask API 异步流
# ============================================================
app = Flask(__name__)

@app.route("/ask_stream", methods=["POST"])
def ask_stream():
    user_input = request.json.get("text", "")

    async def event_stream():
        async for chunk in app_graph.astream({
            "user_input": user_input,
            "city": None,
            "temperature": None,
            "advice": None
        }):
            # 每个 chunk 是当前节点返回的数据
            yield f"data: {json.dumps(chunk)}\n\n"
            await asyncio.sleep(0.05)

    # Flask 支持 SSE (Server-Sent Events)
    return Response(event_stream(), mimetype="text/event-stream")

# ============================================================
# 6. 启动 Flask
# ============================================================
if __name__ == "__main__":
    app.run(host="0.0.0.0", port=5000)

使用

  1. 启动 Flask：
python async_weather_demo.py
  1. 测试 SSE 异步流：
curl -N -X POST http://127.0.0.1:5000/ask_stream \
     -H "Content-Type: application/json" \
     -d '{"text":"北京今天冷吗穿什么？"}'

输出示例（逐步返回）：

data: {"city": "北京"}
data: {"temperature": 5}
data: {"advice": "天气很冷，建议穿厚羽绒服、围巾、手套。"}

前端可以边接收边显示，用户体验更像聊天机器人实时响应。

3.区别

同步调用：invoke()

用法

result = app_graph.invoke(state)

特点

  1. 一步执行完成

    • StateGraph 内所有节点按图的顺序执行完毕后才返回结果。
    • 返回的是最终状态（State 字典或对象）。

  2. 阻塞调用

    • Python 会等待整个图跑完才能继续下一行代码。
    • 如果中间有耗时操作（比如 API 调用、数据库查询），调用线程会被阻塞。

  3. 结果一次性返回

    • 无法看到节点执行的中间状态。
    • 对于长流程或多步骤任务，用户必须等待所有节点完成才能获取响应。

示例

result = app_graph.invoke({
    "user_input": "北京今天冷吗穿什么？",
    "city": None,
    "temperature": None,
    "advice": None
})
print(result)

输出示例：

{
  "city": "北京",
  "temperature": 5,
  "advice": "天气很冷，建议穿厚羽绒服、围巾、手套。"
}

优点：简单直接，适合流程短、响应快的场景。
缺点：无法逐步显示执行过程，耗时长时用户体验差。

异步流调用：astream()

用法

async for chunk in app_graph.astream(state):
    print(chunk)

特点

  1. 异步迭代

    • 每个节点执行完成后就会 yield 当前节点的输出。
    • 可以边执行边获取中间结果。

  2. 非阻塞

    • 可以在执行过程中处理其他任务（适合 async/await 场景）。
    • 适合长流程、API 请求或 RAG 查询等耗时操作。

  3. 流式返回中间状态

    • 每个节点的输出都可以立即返回给前端或日志。
    • 适合前端实时显示“进度”或“逐步结果”。

示例

async for chunk in app_graph.astream({
    "user_input": "北京今天冷吗穿什么？",
    "city": None,
    "temperature": None,
    "advice": None
}):
    print(chunk)

输出示例（逐步返回）：

{'city': '北京'}
{'temperature': 5}
{'advice': '天气很冷，建议穿厚羽绒服、围巾、手套。'}

优点：用户体验好，可实现流式响应、进度显示、分步处理。
缺点：写法稍复杂，需要异步支持（async/await），前端也需要支持 SSE 或 WebSocket 才能实时显示。

对比总结

特性 invoke()（同步） astream()（异步流）
执行方式 阻塞同步 异步迭代（逐步返回）
返回结果 最终状态一次性返回 节点输出逐步返回
适用场景 快速流程，节点少，无长耗时操作 长流程、多节点、耗时操作（API、RAG、LLM）
用户体验 等待所有节点完成才有结果 可实时看到中间执行状态
前端显示 一次性渲染 可流式渲染（SSE、WebSocket）

什么时候选哪个

  • 同步 invoke()

    • 简单场景、节点少、无耗时操作
    • 例如：本地计算、简单逻辑判断

  • 异步流 astream()

    • 流程长、节点多、含网络请求、RAG 或 LLM 调用
    • 需要实时显示中间结果给用户
    • 例如：聊天机器人、问答系统、流程审批、实时推荐
# 服务器 # 人工智能 # 机器学习
Logo
AI Agent技术社区

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

更多推荐

cover

制造业AI模型场景泛化迭代升级：从单点算法到企业级智能体协同的技术演进路径与方案测评

avatar AI Agent技术社区
cover

快消AI Agent落地：从有没有AI到AI能不能产出可考核结论

avatar AI Agent技术社区
cover

企业如何管理Token消耗？

avatar AI Agent技术社区