langgraph-tutorial/README.md

@@ -1,67 +0,0 @@
-# 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

@@ -1,149 +0,0 @@
-"""
-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

@@ -1,105 +0,0 @@
-"""
-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

@@ -1,48 +0,0 @@
-"""
-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

@@ -1,144 +0,0 @@
-"""
-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']}")