commit 2c0e1ca15917c4eb257855bd6313a75ae9f85193 Author: redbotu Date: Mon May 25 21:33:13 2026 +0800 创建demo diff --git a/langgraph-tutorial/README.md b/langgraph-tutorial/README.md new file mode 100644 index 0000000..3943742 --- /dev/null +++ b/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 +``` \ No newline at end of file diff --git a/langgraph-tutorial/agent.py b/langgraph-tutorial/agent.py new file mode 100644 index 0000000..d93e39d --- /dev/null +++ b/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}") \ No newline at end of file diff --git a/langgraph-tutorial/conditional.py b/langgraph-tutorial/conditional.py new file mode 100644 index 0000000..c649c0c --- /dev/null +++ b/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']}") \ No newline at end of file diff --git a/langgraph-tutorial/hello.py b/langgraph-tutorial/hello.py new file mode 100644 index 0000000..413a3c4 --- /dev/null +++ b/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']) \ No newline at end of file diff --git a/langgraph-tutorial/loop.py b/langgraph-tutorial/loop.py new file mode 100644 index 0000000..63937b7 --- /dev/null +++ b/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']}") \ No newline at end of file