使用 LangGraph 构建一个Ai Workflow

from typing import Annotated, TypedDict
from operator import add

class State(TypedDict): # 通常定义为 TypedDict 或 Pydantic 模型。
    messages: Annotated[list, add]  # 消息列表自动追加
    current_plan: str
    approved: bool

def planner_node(state: State):
    # 读取当前状态
    user_input = state["messages"][-1].content

    # 调用 LLM 生成计划
    plan = llm.invoke(f"Plan for: {user_input}")

    # 返回“建议”：只更新 current_plan
    return {"current_plan": plan.content}

def route_plan(state: State):
    if "urgent" in state["plan"]:
        return "fast_executor"
    else:
        return "standard_executor"

workflow.add_conditional_edges("planner", route_plan, {
    "fast_executor": "fast_executor",
    "standard_executor": "standard_executor"
})

from langgraph.graph import StateGraph, START, END

workflow = StateGraph(State)
workflow.add_node("planner", planner_node)
workflow.add_node("executor", executor_node)
workflow.add_edge(START, "planner")
workflow.add_edge("planner", "executor")
workflow.add_edge("executor", END)

app = workflow.compile()

from langgraph.checkpoint.memory import MemorySaver

checkpointer = MemorySaver()
app = workflow.compile(checkpointer=checkpointer)

## 后续可通过 config={"configurable": {"thread_id": "123"}} 追踪会话
result = app.invoke({"messages": [...]}, config={"configurable": {"thread_id": "123"}})

from langgraph.graph import StateGraph, END

## 定义状态结构
class State(TypedDict):
    messages: Annotated[list, operator.add]

## 创建图
workflow = StateGraph(State)

## 添加节点
workflow.add_node("agent", agent_node)
workflow.add_node("tool", tool_node)

## 设置入口和边
workflow.set_entry_point("agent")
workflow.add_edge("agent", "tool")
workflow.add_edge("tool", END)

## 编译并运行
app = workflow.compile()
result = app.invoke({"messages": ["Hello!"]}

uv init langgraph_test
cd langgraph_test
uv venv
source .venv/bin/activate

import os
from typing import TypedDict
from langgraph.graph import StateGraph, END
from langchain_openai import ChatOpenAI
from langchain_core.prompts import ChatPromptTemplate

## 注意：StateGraph 拼写修正（原代码拼错为 StateGragh）
from langgraph.graph import StateGraph  # 正确导入

OPENAI_API_KEY = "<你的 api key>"
OPENAI_API_BASE = "https://dashscope.aliyuncs.com/compatible-mode/v1"

## ---- Step 1: 定义状态（State） ----
class WorkflowState(TypedDict):
    user_input: str
    processed_input: str
    intent: str
    retrieved_data: str
    final_response: str

## ---- Step 2: 初始化 LLM ----
llm = ChatOpenAI(
    model="deepseek-v3",
    api_key=OPENAI_API_KEY,
    base_url=OPENAI_API_BASE,
    temperature=0.1,
)

## ---- Step 3: 定义各个节点函数（增加 print 输出）----

def preprocess_input(state: WorkflowState) -> dict:
    raw = state["user_input"]
    processed = raw.strip().lower()
    print(f"🔍 [preprocess] 输入: '{raw}' → 输出: '{processed}'")
    return {"processed_input": processed}

def detect_intent(state: WorkflowState) -> dict:
    from langchain_core.messages import HumanMessage
    prompt = ChatPromptTemplate.from_messages([
        ("system", "你是一个意图分类器。请根据用户输入判断其意图，只输出一个词：'weather'、'news' 或 'other'。"),
        ("human", "{input}")
    ])
    chain = prompt | llm
    response = chain.invoke({"input": state["processed_input"]})
    intent = response.content.strip().lower()
    if intent not in ["weather", "news"]:
        intent = "other"
    print(f"🧠 [detect_intent] 输入: '{state['processed_input']}' → 意图: '{intent}'")
    return {"intent": intent}

def retrieve_data(state: WorkflowState) -> dict:
    intent = state["intent"]
    if intent == "weather":
        data = "今天北京晴，气温5°C。"
    elif intent == "news":
        data = "今日头条：AI技术取得新突破。"
    else:
        data = "无法提供相关信息。"
    print(f"📦 [retrieve] 意图: '{intent}' → 检索数据: '{data}'")
    return {"retrieved_data": data}

def generate_response(state: WorkflowState) -> dict:
    prompt = ChatPromptTemplate.from_messages([
        ("system", "你是一个有帮助的助手，请根据提供的信息生成简洁友好的回答。"),
        ("human", "用户问题：{question}\n相关信息：{info}")
    ])
    chain = prompt | llm
    response = chain.invoke({
        "question": state["user_input"],
        "info": state["retrieved_data"]
    })
    print(f"💬 [respond] 生成最终回复: '{response.content}'")
    return {"final_response": response.content}

## ---- Step 4: 构建图 ----
workflow = StateGraph(WorkflowState)

workflow.add_node("preprocess", preprocess_input)
workflow.add_node("detect_intent", detect_intent)
workflow.add_node("retrieve", retrieve_data)
workflow.add_node("respond", generate_response)

workflow.set_entry_point("preprocess")
workflow.add_edge("preprocess", "detect_intent")
workflow.add_edge("detect_intent", "retrieve")
workflow.add_edge("retrieve", "respond")
workflow.add_edge("respond", END)

app = workflow.compile()

## ---- Step 5: 运行并观察流程 ----
if __name__ == "__main__":
    inputs = {"user_input": "今天天气怎么样？"}

    print("🚀 开始执行工作流...\n")

    # 方法1：使用 stream() 查看每一步的输出（推荐！）
    for step in app.stream(inputs):
        print("\n📌 当前步骤状态更新:")
        for k, v in step.items():
            print(f"  {k}: {v}")
        print("-" * 50)

    # 如果你只想看最终结果（可选）
    # result = app.invoke(inputs)
    # print("\n✅ 最终结果:", result["final_response"])

🚀 开始执行工作流...

🔍 [preprocess] 输入: '今天天气怎么样？' → 输出: '今天天气怎么样？'

📌 当前步骤状态更新:
  preprocess: {'processed_input': '今天天气怎么样？'}
--------------------------------------------------
🧠 [detect_intent] 输入: '今天天气怎么样？' → 意图: 'weather'

📌 当前步骤状态更新:
  detect_intent: {'intent': 'weather'}
--------------------------------------------------
📦 [retrieve] 意图: 'weather' → 检索数据: '今天北京晴，气温5°C。'

📌 当前步骤状态更新:
  retrieve: {'retrieved_data': '今天北京晴，气温5°C。'}
--------------------------------------------------
💬 [respond] 生成最终回复: '今天北京天气晴朗，气温5°C，适合外出但要注意保暖哦~'

📌 当前步骤状态更新:
  respond: {'final_response': '今天北京天气晴朗，气温5°C，适合外出但要注意保暖哦~'}