基于LangGraph搭建故障根因分析平台思考及总结

from typing import TypedDict
from langgraph.graph import StateGraph, START, END

# --- 第一步：定义状态 (State) ---
# 这是图的"记忆"，所有节点都从这里读取数据，并将结果写回这里
class State(TypedDict):
    content: str

# --- 第二步：定义节点 (Nodes) ---
# 节点就是普通的 Python 函数。它们接收当前 State，返回要更新的字段。

def node_hello(state: State):
    print("🤖 节点 1 正在运行: 添加 Hello")
    # 返回字典会更新 State 中的对应键
    return {"content": state["content"] + " Hello"}

def node_world(state: State):
    print("🌍 节点 2 正在运行: 添加 World")
    return {"content": state["content"] + " World!"}

# --- 第三步：构建图 (Graph Construction) ---

# 1. 初始化图
builder = StateGraph(State)

# 2. 添加节点
builder.add_node("say_hello", node_hello)
builder.add_node("say_world", node_world)

# 3. 添加边 (Edges) - 定义流转逻辑
# START 是图的入口
builder.add_edge(START, "say_hello")     # 开始 -> 节点1
builder.add_edge("say_hello", "say_world") # 节点1 -> 节点2
builder.add_edge("say_world", END)         # 节点2 -> 结束

# 4. 编译图 (Compile)
# 编译将图结构转换为可运行的 Runnable
graph = builder.compile()

# --- 第四步：运行 (Execution) ---
# invoke 需要传入初始状态
initial_state = {"content": "Start:"}
print(f"🏁 初始状态: {initial_state}")

result = graph.invoke(initial_state)

print("-" * 30)
print(f"✅ 最终结果: {result['content']}")

import random
from typing import TypedDict, Literal
from langgraph.graph import StateGraph, START, END

# --- 1. 定义状态 ---
class State(TypedDict):
    face: str    # 记录硬币是 "heads" 还是 "tails"
    message: str # 最终的消息

# --- 2. 定义节点 (Nodes) ---

def node_toss_coin(state: State):
    # 模拟掷硬币逻辑
    result = random.choice(["heads", "tails"])
    print(f"🎲 掷出了: {result}")
    # 更新状态中的 face
    return {"face": result}

def node_winner(state: State):
    print("🏆 走到了赢家节点")
    return {"message": "恭喜！是正面，你赢了！"}

def node_loser(state: State):
    print("😢 走到了输家节点")
    return {"message": "遗憾，是反面，再接再厉。"}

# --- 3. 定义条件逻辑 (The Router) ---
# 这是一个特殊的函数，它不修改状态，只负责"读"状态并返回"路标"
def decide_next_step(state: State) -> Literal["goto_winner", "goto_loser"]:
    if state["face"] == "heads":
        return "goto_winner" # 返回一个自定义的字符串标识
    else:
        return "goto_loser"

# --- 4. 构建图 ---
builder = StateGraph(State)

# 添加节点
builder.add_node("toss", node_toss_coin)
builder.add_node("win_path", node_winner)
builder.add_node("lose_path", node_loser)

# 添加普通边 (START -> toss)
builder.add_edge(START, "toss")

# 添加条件边 (核心部分!)
# 语法: add_conditional_edges(来源节点, 路由函数, 路径映射字典)
builder.add_conditional_edges(
    "toss",              # 从哪个节点出来后开始判断？
    decide_next_step,    # 执行哪个判断函数？
    {                    # 路由函数的返回值 -> 实际节点名的映射
        "goto_winner": "win_path",
        "goto_loser":  "lose_path"
    }
)

# 两个分支最终都汇聚到 END
builder.add_edge("win_path", END)
builder.add_edge("lose_path", END)

# 编译与运行
graph = builder.compile()

print("--- 开始运行 ---")
result = graph.invoke({"face": "", "message": ""})
print("-" * 30)
print(f"🏁 最终结果: {result['message']}")

from typing import TypedDict, Literal
from langgraph.graph import StateGraph, START, END


# --- 1. 定义状态 ---
class State(TypedDict):
    draft: str  # 当前的草稿内容
    count: int  # 计数器（防止死循环的关键！）


# --- 2. 定义节点 ---

def node_writer(state: State):
    current_count = state["count"] + 1
    print(f"✍️ [第 {current_count} 次迭代] 作家正在写作...")

    # 模拟：每次迭代都在草稿后追加一点内容
    new_content = state["draft"] + " 写了一点内容。"

    return {
        "draft": new_content,
        "count": current_count
    }


# --- 3. 定义循环逻辑 (Condition) ---
def check_quality(state: State) -> Literal["retry", "finish"]:
    # 逻辑 A: 强制退出机制 (防止死循环)
    if state["count"] >= 3:
        print("🛑 达到最大重试次数，强制结束。")
        return "finish"

    # 逻辑 B: 质量检查 (比如模拟长度检查)
    if len(state["draft"]) > 20:
        print("✅ 字数达标，通过！")
        return "finish"

    # 逻辑 C: 否则重试
    print("🔙 字数太少，退回重写...")
    return "retry"


# --- 4. 构建图 ---
builder = StateGraph(State)

builder.add_node("writer", node_writer)

# 入口 -> 作家
builder.add_edge(START, "writer")

# 关键点：条件边的目标可以是它自己，或者之前的节点！
builder.add_conditional_edges(
    "writer",  # 从 writer 出来
    check_quality,  # 检查质量
    {
        "retry": "writer",  # <--- 这里形成了循环！指向了自己
        "finish": END
    }
)

graph = builder.compile()

# --- 5. 运行 ---
print("--- 开始循环测试 ---")
# 初始给一个空草稿，计数器为0
initial_state = {"draft": "", "count": 0}
result = graph.invoke(initial_state)

print("-" * 30)
print(f"📝 最终草稿: {result['draft']}")
print(f"🔢 最终尝试次数: {result['count']}")

from typing import TypedDict, Annotated
from langgraph.graph import StateGraph, START, END
from langgraph.checkpoint.memory import MemorySaver # <--- 1. 引入存档器

# 定义状态：使用 add_messages 类似的逻辑，这里简化为字符串拼接
class State(TypedDict):
    # 关键修复：使用 Annotated 定义 reducer 函数，用于合并新旧状态
    # 当 invoke 传入新输入时，reducer 会调用这个函数来合并，而不是直接替换
    conversation: Annotated[str, lambda x, y: x + y]

def node_chatbot(state: State):
    # 模拟处理：把新输入追加到历史记录中
    # 注意：由于 conversation 已经有 reducer，新输入在 invoke 时已经自动追加到历史状态中
    # 节点不需要返回 conversation，因为：
    # 1. invoke 时传入的新输入已经通过 reducer 合并到恢复的状态中
    # 2. 如果节点返回 conversation，会再次通过 reducer 合并，导致重复
    # 如果节点需要添加新内容（如 AI 回复），应该只返回新内容，让 reducer 自动合并
    # 这里节点不做任何更新，所以返回空字典
    return {}

# --- 构建图 ---
builder = StateGraph(State)
builder.add_node("chatbot", node_chatbot)
builder.add_edge(START, "chatbot")
builder.add_edge("chatbot", END)

# --- 2. 初始化存档器并编译 ---
memory = MemorySaver() # 在内存中保存 (生产环境通常用 PostgresSaver)
graph = builder.compile(checkpointer=memory)

# --- 3. 运行测试 ---

# 设定一个 ID (这就好比你的用户 ID)
config = {"configurable": {"thread_id": "user_123"}}

print("--- 第一轮对话 ---")
# 用户说 "Hi"
input_1 = {"conversation": "User: Hi! | "}
# 传入 config，让图知道要存到哪个 ID 下
graph.invoke(input_1, config=config)

# 查看一下当前记忆 (这步不是必须的，仅供调试)
snapshot = graph.get_state(config)
print(f"💾 当前存档: {snapshot.values['conversation']}")

print("\n--- 第二轮对话 (模拟第二天) ---")
# 注意：我们这里不需要把 "User: Hi" 再次传进去，图自己会去 thread_id 读取！
input_2 = {"conversation": "User: My name is Jim. | "} # 只传入新内容
result = graph.invoke(input_2, config=config)

print(f"✅ 最终完整记忆: {result['conversation']}")

from typing import TypedDict
from langgraph.graph import StateGraph, START, END
from langgraph.checkpoint.memory import MemorySaver

# --- 1. 定义状态 ---
class State(TypedDict):
    amount: int
    approved: bool
    status: str

# --- 2. 定义节点 ---

def step_1_request(state: State):
    print(f"🤖 (自动) 步骤 1: 申请转账 ${state['amount']}")
    # 默认 approved 是 False
    return {"status": "pending_approval"}

def step_2_execute(state: State):
    # 这个节点会在"断点"之后运行
    if state["approved"]:
        print(f"💸 (自动) 步骤 2: 转账成功！金额: ${state['amount']}")
        return {"status": "success"}
    else:
        print(f"🚫 (自动) 步骤 2: 转账被拒绝。")
        return {"status": "rejected"}

# --- 3. 构建图 ---
builder = StateGraph(State)
builder.add_node("request", step_1_request)
builder.add_node("execute", step_2_execute)

builder.add_edge(START, "request")
builder.add_edge("request", "execute")
builder.add_edge("execute", END)

# --- 4. 关键设置：添加 Memory 和 断点 ---
memory = MemorySaver()

# interrupt_before=["execute"]:
# 意思是：在进入 "execute" 节点之前，暂停运行！
graph = builder.compile(
    checkpointer=memory,
    interrupt_before=["execute"]
)

# --- 5. 运行模拟 ---

config = {"configurable": {"thread_id": "tx_999"}}

print("--- 🔴 阶段 1: 启动任务 ---")
# 初始请求转账 1000 元
graph.invoke({"amount": 1000, "approved": False}, config=config)

# 此时，程序已经暂停了。我们可以检查一下当前状态。
snapshot = graph.get_state(config)
print(f"⏸️ 程序已暂停。下一步要去: {snapshot.next}")
print(f"👀 当前状态: {snapshot.values}")

print("\n--- 🧑‍💻 阶段 2: 人类介入 (审批) ---")
# 假设人类看了眼，决定批准。我们使用 update_state 修改内存中的数据。
graph.update_state(config, {"approved": True})
print("✅ 人类已将状态修改为 Approved。")

print("\n--- 🟢 阶段 3: 恢复运行 ---")
# 传入 None 表示"继续执行下一步"，它会读取刚才更新过的状态
graph.invoke(None, config=config)