Merge branch 'main' of https://git.redbotu.com/dark/langgraph_learing

langgraph-tutorial/README.md

@@ -0,0 +1,67 @@
+# LangGraph 学习教程
+
+## 课程大纲
+
+### ✅ 第 1 步：核心理念
+- LangGraph 是基于图的 AI 应用框架
+- 支持有状态、循环、分支的智能体构建
+
+### ✅ 第 2 步：Hello World
+- 文件: `hello.py`
+- 学习了：State、Node、Edge 三大核心概念
+
+### ✅ 第 3 步：条件边
+- 文件: `conditional.py`
+- 学习了：让图学会做决定，根据输入走不同路径
+
+### ✅ 第 4 步：循环 + 智能体
+- 文件: `loop.py` - 循环演示（尝试-评估-重试）
+- 文件: `agent.py` - AI 智能体（思考-行动-回答）
+
+## 核心概念总结
+
+```
+LangGraph 图 = State(状态) + Node(节点) + Edge(边)
+
+State: TypedDict，节点间传递的数据包
+Node: 函数，接收状态并返回更新
+Edge: 连接节点，定义流程走向
+  - 固定边: add_edge()
+  - 条件边: add_conditional_edges()
+```
+
+## 下一步学习
+
+### 第 5 步：真实 LLM 智能体
+- 接入 OpenAI API
+- 构建 ReAct 模式智能体
+- 添加工具调用能力
+
+### 第 6 步：高级特性
+- 检查点机制（暂停/恢复）
+- 子图（嵌套图）
+- 并行执行
+- 流式输出
+
+## 运行示例
+
+```bash
+# 设置编码
+$env:PYTHONIOENCODING="utf-8"
+
+# 运行各个示例
+python hello.py      # Hello World
+python conditional.py # 条件边
+python loop.py       # 循环
+python agent.py      # AI 智能体
+```
+
+## 接入真实 LLM
+
+```powershell
+# 设置 API Key
+$env:OPENAI_API_KEY = "sk-..."
+
+# 运行智能体示例
+python agent.py
+```

langgraph-tutorial/agent.py

@@ -0,0 +1,149 @@
+"""
+LangGraph 实战 - 接入 OpenAI LLM 的智能体
+构建一个能"思考-行动-反思"的真实 AI 智能体
+"""
+import os
+from langgraph.graph import StateGraph, START, END
+from typing import TypedDict, Annotated
+from functools import reduce
+
+# 检查 API Key
+OPENAI_KEY = os.environ.get("OPENAI_API_KEY", "")
+if not OPENAI_KEY:
+    print("请先设置 OPENAI_API_KEY:")
+    print("  $env:OPENAI_API_KEY = 'sk-...'  # PowerShell")
+    print()
+    print("暂时使用模拟模式演示...")
+    USE_MOCK = True
+else:
+    USE_MOCK = False
+
+# 1️⃣ 定义状态
+class AgentState(TypedDict):
+    query: str
+    thought: str
+    action: str
+    observation: str
+    answer: str
+    max_iterations: int
+    iteration: int
+
+# 2️⃣ 模拟 LLM 调用
+def mock_llm(prompt: str) -> str:
+    """模拟 LLM 响应"""
+    responses = {
+        "思考": "我需要先分析问题，然后给出解答。",
+        "行动": "搜索相关信息并整理答案。",
+        "回答": "根据分析，LangGraph 是一个用于构建有状态 AI 应用的框架。",
+    }
+    for key, value in responses.items():
+        if key in prompt:
+            return value
+    return "这是一个有趣的问题。"
+
+def real_llm(prompt: str) -> str:
+    """真实 LLM 调用"""
+    from openai import OpenAI
+    client = OpenAI()
+    response = client.chat.completions.create(
+        model="gpt-4o-mini",
+        messages=[{"role": "user", "content": prompt}],
+        max_tokens=200,
+        temperature=0.7,
+    )
+    return response.choices[0].message.content
+
+# 3️⃣ 定义节点
+def think_node(state: AgentState):
+    """思考节点 - 分析当前情况"""
+    state = state.copy()
+    state['iteration'] += 1
+    print(f"\n{'='*40}")
+    print(f"[思考] 第 {state['iteration']}/{state['max_iterations']} 轮")
+    
+    if USE_MOCK:
+        state['thought'] = mock_llm(f"思考: {state['query']}")
+    else:
+        state['thought'] = real_llm(f"思考: {state['query']}")
+    
+    print(f"  想法: {state['thought']}")
+    return state
+
+def act_node(state: AgentState):
+    """行动节点 - 执行操作"""
+    state = state.copy()
+    print(f"[行动] 执行操作...")
+    
+    if USE_MOCK:
+        state['action'] = mock_llm("行动")
+        state['observation'] = f"观察到: {state['action']}的结果"
+    else:
+        state['action'] = real_llm(f"针对问题 '{state['query']}' 我该采取什么行动？")
+        state['observation'] = f"执行了: {state['action']}"
+    
+    print(f"  行动: {state['action']}")
+    print(f"  观察: {state['observation']}")
+    return state
+
+def answer_node(state: AgentState):
+    """回答节点 - 给出最终答案"""
+    state = state.copy()
+    print(f"\n[回答] 生成最终答案...")
+    
+    if USE_MOCK:
+        state['answer'] = mock_llm("回答")
+    else:
+        prompt = f"""基于以下信息回答问题:
+问题: {state['query']}
+思考过程: {state['thought']}
+行动: {state['action']}
+观察: {state['observation']}
+请给出简洁的答案:"""
+        state['answer'] = real_llm(prompt)
+    
+    print(f"  答案: {state['answer']}")
+    return state
+
+# 4️⃣ 路由函数
+def should_continue(state: AgentState):
+    """决定是否继续迭代"""
+    if state['iteration'] >= state['max_iterations']:
+        return "answer"
+    observation = state.get('observation', '')
+    if observation and len(observation) > 10:
+        return "answer"
+    return "act"
+
+# 5️⃣ 构建图
+graph = StateGraph(AgentState)
+graph.add_node("think", think_node)
+graph.add_node("act", act_node)
+graph.add_node("answer", answer_node)
+
+# 连接
+graph.add_edge(START, "think")
+graph.add_conditional_edges("think", should_continue, {"act": "act", "answer": "answer"})
+graph.add_edge("act", "answer")
+graph.add_edge("answer", END)
+
+# 编译
+app = graph.compile()
+
+# 6️⃣ 运行
+print("=" * 40)
+print("LangGraph AI 智能体演示")
+print("=" * 40)
+
+result = app.invoke({
+    "query": "什么是 LangGraph？",
+    "thought": "",
+    "action": "",
+    "observation": "",
+    "answer": "",
+    "max_iterations": 2,
+    "iteration": 0,
+})
+
+print(f"\n{'='*40}")
+print(f"最终答案: {result['answer']}")
+print(f"{'='*40}")

langgraph-tutorial/conditional.py

@@ -0,0 +1,105 @@
+"""
+LangGraph 条件边 - 让图学会做决定
+"""
+from langgraph.graph import StateGraph, START, END
+from typing import TypedDict
+
+# 1️⃣ 定义状态
+class GraphState(TypedDict):
+    message: str
+    user_input: str
+
+# 2️⃣ 定义节点
+def receive_node(state: GraphState):
+    """接收节点 - 分类用户输入"""
+    text = state['user_input'].strip().lower()
+    print(f"[接收] 用户说: {state['user_input']}")
+    
+    # 简单分类
+    if any(word in text for word in ['你好', 'hello', 'hi', 'hey']):
+        state['message'] = "问候"
+    elif '?' in text or '？' in text or '怎么' in text or '什么' in text:
+        state['message'] = "问题"
+    else:
+        state['message'] = "闲聊"
+    
+    print(f"[接收] 分类为: {state['message']}")
+    return state
+
+def respond_greeting(state: GraphState):
+    """回应问候"""
+    state['message'] = "你好呀！有什么可以帮你的吗？"
+    print(f"[问候回应] {state['message']}")
+    return state
+
+def answer_question(state: GraphState):
+    """回答问题"""
+    state['message'] = "这是一个好问题！让我想想... (这里可以接入 LLM)"
+    print(f"[问题回答] {state['message']}")
+    return state
+
+def chat_casual(state: GraphState):
+    """闲聊"""
+    state['message'] = "哈哈，聊得开心！"
+    print(f"[闲聊] {state['message']}")
+    return state
+
+# 3️⃣ 条件路由函数 - 这是关键！
+def route_input(state: GraphState):
+    """根据分类决定下一步走哪个节点"""
+    category = state['message']
+    print(f"[路由] 决定走向: {category}")
+    
+    if category == "问候":
+        return "respond_greeting"
+    elif category == "问题":
+        return "answer_question"
+    else:
+        return "chat_casual"
+
+# 4️⃣ 构建图
+graph = StateGraph(GraphState)
+
+# 添加所有节点
+graph.add_node("receive", receive_node)
+graph.add_node("respond_greeting", respond_greeting)
+graph.add_node("answer_question", answer_question)
+graph.add_node("chat_casual", chat_casual)
+
+# 添加边
+graph.add_edge(START, "receive")
+
+# 条件边！根据路由函数返回值决定走向
+graph.add_conditional_edges(
+    "receive",           # 从接收节点出发
+    route_input,         # 用这个函数做判断
+    {
+        "respond_greeting": "respond_greeting",
+        "answer_question": "answer_question",
+        "chat_casual": "chat_casual",
+    }
+)
+
+# 所有分支最后都汇聚到 END
+graph.add_edge("respond_greeting", END)
+graph.add_edge("answer_question", END)
+graph.add_edge("chat_casual", END)
+
+# 编译
+app = graph.compile()
+
+# 5️⃣ 测试三种情况
+test_cases = [
+    "你好！",
+    "什么是 LangGraph？",
+    "今天天气不错",
+]
+
+print("=" * 50)
+print("条件边演示 - 图会根据输入走不同路径")
+print("=" * 50)
+
+for i, text in enumerate(test_cases, 1):
+    print(f"\n--- 测试 {i}: {text} ---")
+    result = app.invoke({"user_input": text, "message": ""})
+    print(f"  最终输出: {result['message']}")

langgraph-tutorial/hello.py

@@ -0,0 +1,48 @@
+"""
+LangGraph Hello World - 最简单的图
+"""
+from langgraph.graph import StateGraph, START, END
+from typing import TypedDict, Annotated
+import operator
+
+# 1️⃣ 定义状态 - 这是节点之间传递的数据结构
+class GraphState(TypedDict):
+    """图的状态 - 节点之间共享的数据"""
+    message: str  # 一条消息
+
+# 2️⃣ 定义节点 - 每个节点是一个函数，接收状态并返回更新
+def greet_node(state: GraphState):
+    """问候节点"""
+    print(f"[问候节点] 收到: {state['message']}")
+    return {"message": "你好！我是 LangGraph 智能体 👋"}
+
+def process_node(state: GraphState):
+    """处理节点"""
+    print(f"[处理节点] 收到: {state['message']}")
+    return {"message": state['message'] + " 很高兴见到你！"}
+
+# 3️⃣ 构建图
+graph = StateGraph(GraphState)
+
+# 添加节点
+graph.add_node("greet", greet_node)
+graph.add_node("process", process_node)
+
+# 添加边 - 定义流程走向
+# START -> greet -> process -> END
+graph.add_edge(START, "greet")       # 开始 -> 问候节点
+graph.add_edge("greet", "process")   # 问候 -> 处理节点
+graph.add_edge("process", END)       # 处理 -> 结束
+
+# 编译图
+app = graph.compile()
+
+# 4️⃣ 运行！
+print("=" * 40)
+print("🚀 运行 LangGraph Hello World")
+print("=" * 40)
+
+result = app.invoke({"message": "Hi!"})
+
+print()
+print("最终结果:", result['message'])

langgraph-tutorial/loop.py

@@ -0,0 +1,144 @@
+"""
+LangGraph 循环 - 智能体的核心能力
+构建一个简单的"尝试-评估-重试"智能体
+"""
+from langgraph.graph import StateGraph, START, END
+from typing import TypedDict
+
+# 1️⃣ 定义状态 - 累积信息
+class AgentState(TypedDict):
+    task: str          # 任务
+    attempt: int       # 尝试次数
+    result: str        # 当前结果
+    quality: str       # 质量评估: "good" / "needs_improvement" / "bad"
+    feedback: str      # 改进建议
+
+# 2️⃣ 定义节点
+def plan_node(state: AgentState):
+    """规划节点 - 制定初始方案"""
+    print(f"\n[规划] 任务: {state['task']}")
+    print(f"[规划] 制定初始方案...")
+    
+    # 模拟规划逻辑
+    if "诗" in state['task']:
+        state['result'] = "床前明月光"
+        state['quality'] = "needs_improvement"
+    elif "代码" in state['task'] or "code" in state['task'].lower():
+        state['result'] = "print(hello)"  # 有 bug
+        state['quality'] = "bad"
+    else:
+        state['result'] = "这是一个初步方案"
+        state['quality'] = "needs_improvement"
+    
+    state['attempt'] = 1
+    print(f"[规划] 第 {state['attempt']} 次尝试: {state['result']}")
+    return state
+
+def execute_node(state: AgentState):
+    """执行节点 - 根据反馈改进"""
+    state['attempt'] += 1
+    print(f"\n[执行] 第 {state['attempt']} 次尝试...")
+    print(f"[执行] 收到反馈: {state['feedback']}")
+    
+    # 模拟根据反馈改进
+    if "诗" in state['task']:
+        if state['attempt'] == 2:
+            state['result'] = "床前明月光，疑是地上霜"
+        else:
+            state['result'] = "床前明月光，疑是地上霜。举头望明月，低头思故乡。"
+    elif "代码" in state['task'] or "code" in state['task'].lower():
+        if state['attempt'] == 2:
+            state['result'] = "print('hello')"  # 修复了
+        else:
+            state['result'] = "print('hello, world!')  # 完美!"
+    
+    print(f"[执行] 改进后: {state['result']}")
+    return state
+
+def evaluate_node(state: AgentState):
+    """评估节点 - 判断质量"""
+    print(f"\n[评估] 检查第 {state['attempt']} 次结果...")
+    
+    # 模拟评估逻辑
+    if state['attempt'] >= 3:
+        state['quality'] = "good"
+        state['feedback'] = "质量满意，通过！"
+        print(f"[评估] 结果: {state['quality']} - {state['feedback']}")
+    else:
+        if "诗" in state['task']:
+            state['feedback'] = "内容太短，需要更完整"
+        elif "代码" in state['task'] or "code" in state['task'].lower():
+            state['feedback'] = "代码有语法错误，需要修复"
+        else:
+            state['feedback'] = "需要更详细的内容"
+        state['quality'] = "needs_improvement"
+        print(f"[评估] 结果: {state['quality']} - {state['feedback']}")
+    
+    return state
+
+# 3️⃣ 路由函数 - 决定是继续循环还是结束
+def should_retry(state: AgentState):
+    """⭐ 核心：决定是重试还是结束"""
+    if state['quality'] == "good":
+        return "end"          # 质量够了，结束
+    elif state['attempt'] >= 3:
+        return "end"          # 达到最大重试次数，强制结束
+    else:
+        return "retry"        # 继续改进
+
+# 4️⃣ 构建图
+graph = StateGraph(AgentState)
+
+# 添加节点
+graph.add_node("plan", plan_node)
+graph.add_node("execute", execute_node)
+graph.add_node("evaluate", evaluate_node)
+
+# 添加边 - 注意这里的循环！
+graph.add_edge(START, "plan")           # 开始 -> 规划
+graph.add_edge("plan", "evaluate")      # 规划 -> 评估
+graph.add_conditional_edges(
+    "evaluate",
+    should_retry,
+    {
+        "retry": "execute",   # 需要改进 -> 执行(改进)
+        "end": END,           # 满意 -> 结束
+    }
+)
+graph.add_edge("execute", "evaluate")   # 执行完再评估 -> 形成循环！
+
+# 编译
+app = graph.compile()
+
+# 5️⃣ 测试
+print("=" * 55)
+print("循环演示 - 智能体不断尝试直到满意")
+print("=" * 55)
+
+# 测试 1: 写诗
+print("\n" + "=" * 55)
+print("测试 1: 写一首诗")
+print("=" * 55)
+result = app.invoke({
+    "task": "写一首诗",
+    "attempt": 0,
+    "result": "",
+    "quality": "",
+    "feedback": ""
+})
+print(f"\n最终结果: {result['result']}")
+print(f"尝试次数: {result['attempt']}")
+
+# 测试 2: 写代码
+print("\n" + "=" * 55)
+print("测试 2: 写一段代码")
+print("=" * 55)
+result = app.invoke({
+    "task": "写一段代码",
+    "attempt": 0,
+    "result": "",
+    "quality": "",
+    "feedback": ""
+})
+print(f"\n最终结果: {result['result']}")
+print(f"尝试次数: {result['attempt']}")