From bc1a20caa568a5401cbdf9c7d6c69bab7239d903 Mon Sep 17 00:00:00 2001 From: redbotu Date: Mon, 25 May 2026 22:22:15 +0800 Subject: [PATCH] =?UTF-8?q?docs:=20=E6=B7=BB=E5=8A=A0=20Java=20=E5=BC=80?= =?UTF-8?q?=E5=8F=91=E8=80=85=20Python=20=E8=AF=AD=E6=B3=95=E5=AF=B9?= =?UTF-8?q?=E7=85=A7=E6=B3=A8=E9=87=8A?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- langgraph-tutorial/react_agent.py | 363 +++++++++++++++--------------- 1 file changed, 179 insertions(+), 184 deletions(-) diff --git a/langgraph-tutorial/react_agent.py b/langgraph-tutorial/react_agent.py index b313434..26fbfce 100644 --- a/langgraph-tutorial/react_agent.py +++ b/langgraph-tutorial/react_agent.py @@ -2,107 +2,151 @@ LangGraph 第 5 步：真实 LLM 智能体 ReAct 模式：思考(Reason) - 行动(Act) - 观察(Observe) 配置见 config.py + +【Java 开发者 Python 速查】 +- 没有分号 ; 结尾 +- 用缩进（4空格）表示代码块，不用 {} +- def 定义函数（类似 Java 的方法） +- class 定义类 +- import 导入模块（类似 Java 的 import） +- 变量不用声明类型（但可以用 :str 做类型提示） +- True/False/None（类似 Java 的 true/false/null） +- list 类似 ArrayList，dict 类似 HashMap +- for item in list: 遍历（类似 Java 的 for-each） """ # ============================================================ # 导入模块 # ============================================================ -import sys # 命令行参数 -import requests # HTTP 请求（调用 LLM API） -from langgraph.graph import StateGraph, START, END # LangGraph 核心组件 -from typing import TypedDict # 类型提示，用于定义状态结构 +# Java: import java.util.List; +# Python: import 模块名 +import sys # 命令行参数（类似 Java 的 main 的 String[] args） +import requests # HTTP 请求库（类似 Java 的 OkHttp/RestTemplate） +from langgraph.graph import StateGraph, START, END # 从模块导入特定类 +from typing import TypedDict # 类型提示工具 # ============================================================ # 1. 加载配置 # ============================================================ -# 从 config.py 导入配置变量 -# Python 的 import 语句可以直接导入另一个 .py 文件中的变量 +# Java: 从配置文件读取 +# Python: 直接 import 另一个 .py 文件，像导入类一样 from config import API_KEY, BASE_URL, MODEL, MAX_ITERATIONS, TEMPERATURE -# 检查用户是否修改了配置 +# if 条件判断 +# Java: if (condition) { ... } +# Python: if condition: +# 缩进表示代码块（4个空格） if API_KEY == "sk-...": print("请先在 config.py 中配置 API_KEY") - exit(1) # 退出程序 + exit(1) # 退出程序（类似 Java 的 System.exit(1)） # ============================================================ # 2. 定义状态（State） # ============================================================ """ -TypedDict 是 Python 的类型提示工具，用来定义字典的结构。 -LangGraph 用 State 在节点之间传递数据，每个节点可以读取和修改它。 +TypedDict 是 Python 的类型提示工具。 -为什么用 TypedDict 而不是普通 dict？ -- 编辑器可以提供自动补全 -- 可以检查键名是否正确 -- 代码更清晰，一看就知道有哪些字段 +Java 对比: + // Java + class AgentState { + String question; + List thoughts; + int iteration; + } + + // Python + class AgentState(TypedDict): + question: str # String + thoughts: list # List + iteration: int # int + +注意: Python 的类型提示只是给编辑器看的，运行时不强制检查。 + TypedDict 本质上还是 dict（字典），只是加了类型说明。 """ class AgentState(TypedDict): - question: str # 用户的问题 - thoughts: list # 思考历史（list，记录每轮的思考） - current_thought: str # 当前这轮的思考 - action: str # 要采取的行动（如 "calculator"） - action_param: str # 行动的参数（如 "123*456"） - observation: str # 行动后的观察结果 - final_answer: str # 最终答案 - iteration: int # 当前迭代轮次 - max_iterations: int # 最大迭代次数（防止无限循环） + question: str # String - 用户的问题 + thoughts: list # List - 思考历史 + current_thought: str # String - 当前这轮的思考 + action: str # String - 要采取的行动 + action_param: str # String - 行动的参数 + observation: str # String - 行动后的观察结果 + final_answer: str # String - 最终答案 + iteration: int # int - 当前迭代轮次 + max_iterations: int # int - 最大迭代次数 # ============================================================ # 3. 定义工具（Tools） # ============================================================ """ -工具是智能体可以调用的函数。 -这里定义了两个工具：计算器和知识搜索。 +工具就是 Python 函数，智能体可以调用它们。 -为什么工具是普通函数？ -- LangGraph 中工具就是 Python 函数 -- 智能体决定调用哪个工具、传什么参数 -- 工具执行后返回结果，智能体再根据结果决定下一步 +Java 对比: + // Java: 需要定义接口和实现 + public interface Tool { + String execute(String param); + } + + // Python: 直接就是函数 + def calculator(expression: str) -> str: + ... + +Python 函数定义: + def 函数名(参数: 类型) -> 返回类型: + 函数体（缩进） + return 返回值 """ def calculator(expression: str) -> str: """ 计算器工具 - 参数: expression - 数学表达式字符串，如 "2+3*4" - 返回: 计算结果字符串 - - 为什么用 eval 而不是直接计算？ - - eval 可以解析任意数学表达式 - - 但 eval 有安全风险，所以传了空的 __builtins__ 限制权限 + expression: str - 参数类型提示（String） + -> str - 返回类型提示（String） """ try: # eval() 把字符串当 Python 表达式执行 + # Java: 需要用脚本引擎，Python 直接 eval # {"__builtins__": {}} 是安全限制，防止执行危险代码 result = eval(expression, {"__builtins__": {}}, {}) + + # f-string: Python 的字符串格式化 + # Java: String.format("结果: %s = %s", expression, result) + # Python: f"结果: {expression} = {result}" return f"计算结果: {expression} = {result}" except Exception as e: - # 如果计算出错（如除零、无效表达式），返回错误信息 + # try-catch 语法 + # Java: catch (Exception e) { ... } + # Python: except Exception as e: return f"计算错误: {e}" def search_knowledge(query: str) -> str: """ 知识库搜索工具（模拟） - 为什么用字典模拟而不是真实搜索？ - - 教学目的，先理解流程 - - 真实项目中可以替换为搜索引擎 API、向量数据库等 + Python dict（字典）对比 Java Map: + # Java: Map map = new HashMap<>(); + # map.put("key", "value"); + + # Python: dict = {"key": "value"} """ - # 一个简单的知识库（字典） knowledge = { - "langgraph": "LangGraph 是 LangChain 团队开发的框架，用于构建有状态、基于图的 AI 应用。", - "python": "Python 是一种高级编程语言，广泛用于 AI、Web 开发、数据分析等领域。", - "langchain": "LangChain 是构建 LLM 应用的框架，提供 Prompt 管理、Chain、Agent 等组件。", - "ai": "人工智能 (AI) 是计算机科学的一个分支，致力于创建能执行智能任务的系统。", + "langgraph": "LangGraph 是 LangChain 团队开发的框架...", + "python": "Python 是一种高级编程语言...", + "langchain": "LangChain 是构建 LLM 应用的框架...", + "ai": "人工智能 (AI) 是计算机科学的一个分支...", } - # 遍历知识库，查找匹配的关键词 + + # for 循环遍历字典 + # Java: for (Map.Entry entry : knowledge.entrySet()) + # Python: for key, value in knowledge.items(): for key, value in knowledge.items(): - if key in query.lower(): # 转小写后匹配 + if key in query.lower(): # .lower() 转小写（类似 Java 的 toLowerCase()） return f"搜索到: {value}" return f"未找到关于 '{query}' 的精确信息" -# 把工具放进字典，方便通过名称查找 -# 键是工具名（LLM 输出的），值是函数对象 +# 工具注册表：dict 映射 工具名 -> 函数 +# Java: Map> tools = new HashMap<>(); +# Python: tools = {"name": function} tools = { "calculator": calculator, "search": search_knowledge, @@ -111,71 +155,58 @@ tools = { # ============================================================ # 4. LLM 调用函数 # ============================================================ -""" -为什么不用 OpenAI SDK 而用 requests？ -- 某些 API 网关与 OpenAI SDK 不兼容 -- requests 更灵活，可以直接控制请求格式 -- 原理相同，只是底层 HTTP 调用方式不同 -""" def call_llm(messages, max_tokens=300): """ 调用 LLM API - 参数: - messages: 消息列表，格式为 [{"role": "system/user/assistant", "content": "..."}] - max_tokens: 最大生成 token 数 - - 返回: LLM 生成的文本 + max_tokens=300 是默认参数 + Java: 需要方法重载 + Python: 直接在参数列表中给默认值 """ - # 构建 API 请求 - url = f"{BASE_URL}/chat/completions" # 拼接完整的 API 地址 + url = f"{BASE_URL}/chat/completions" headers = { - "Content-Type": "application/json", # 告诉服务器发送 JSON - "Authorization": f"Bearer {API_KEY}" # API 认证 + "Content-Type": "application/json", + "Authorization": f"Bearer {API_KEY}" } body = { - "model": MODEL, # 使用哪个模型 - "messages": messages, # 对话消息 - "max_tokens": max_tokens, # 最大生成长度 - "temperature": TEMPERATURE, # 创造性（0=确定，1=随机） + "model": MODEL, + "messages": messages, + "max_tokens": max_tokens, + "temperature": TEMPERATURE, } - # 发送 HTTP POST 请求 + # requests.post() 发送 HTTP POST 请求 + # Java: 需要 OkHttp/RestTemplate 几行代码 + # Python: 一行搞定 response = requests.post(url, headers=headers, json=body, timeout=30) - response.raise_for_status() # 如果 HTTP 状态码不是 200，抛出异常 - data = response.json() # 把响应解析为 JSON + response.raise_for_status() # 检查 HTTP 状态码 + data = response.json() # 解析 JSON（类似 Java 的 Jackson/Gson） - # 从 JSON 响应中提取 LLM 生成的文本 - # 响应结构: {"choices": [{"message": {"content": "生成的文本"}}]} + # 访问嵌套 JSON + # Java: data.get("choices").get(0).get("message").get("content") + # Python: data["choices"][0]["message"]["content"] return data["choices"][0]["message"]["content"] # ============================================================ # 5. 定义节点（Nodes） # ============================================================ """ -节点是 LangGraph 图中的处理单元。 -每个节点是一个函数，接收 State，返回更新后的 State。 +LangGraph 节点：接收 State，返回更新后的 State。 为什么每个节点都要 state = state.copy()？ -- Python 字典是可变对象，直接修改会影响原始数据 -- .copy() 创建浅拷贝，确保每个节点操作的是自己的副本 -- 这是 LangGraph 的最佳实践 +- Python 字典是可变对象（mutable） +- 直接修改会影响原始数据 +- .copy() 创建浅拷贝（类似 Java 的 new HashMap<>(original)） """ def think_node(state: AgentState): - """ - 思考节点 - 让 LLM 决定下一步 - - 这是 ReAct 模式的核心： - 1. 把问题和历史发给 LLM - 2. LLM 输出 [思考] 和 [行动] 或 [回答] - 3. 解析 LLM 的输出，提取行动或答案 - """ - state = state.copy() # 创建副本，避免修改原始状态 - state['iteration'] += 1 # 迭代次数 +1 + """思考节点 - 让 LLM 决定下一步""" + state = state.copy() # 创建副本 + state['iteration'] += 1 # 迭代次数 +1（类似 Java 的 state.iteration++） - # 构建系统提示词（System Prompt） - # f-string 中的 {state['iteration']} 会被替换为实际值 + # 多行字符串（三引号） + # Java: "line1\n" + "line2\n" + # Python: """line1\nline2""" system_prompt = f"""你是一个智能助手，可以使用以下工具: 1. calculator - 数学计算，参数是数学表达式如 "2+3*4" 2. search - 搜索知识，参数是搜索关键词 @@ -193,13 +224,16 @@ def think_node(state: AgentState): [思考] 我已经知道答案了 [回答] 你的最终答案""" - # 构建消息列表（Messages） - # OpenAI API 的消息格式：角色 + 内容 + # 构建消息列表 + # Java: List> messages = new ArrayList<>(); + # messages.add(Map.of("role", "system", "content", prompt)); + # Python: messages = [{"role": "system", "content": prompt}] messages = [{"role": "system", "content": system_prompt}] messages.append({"role": "user", "content": state['question']}) - # 如果有上一次的观察结果，也发给 LLM - # 这样 LLM 可以根据工具返回的结果做下一步决定 + # .get() 安全访问 + # Java: String obs = state.get("observation"); + # Python: obs = state.get('observation') # 不存在返回 None if state.get('observation'): messages.append({"role": "assistant", "content": f"[观察] {state['observation']}"}) @@ -209,48 +243,51 @@ def think_node(state: AgentState): # 保存思考历史 state['current_thought'] = thought_text state['thoughts'] = state.get('thoughts', []) + [thought_text] + # .get('thoughts', []) - 如果不存在返回空列表 [] # 打印调试信息 - print(f"\n{'='*50}") # {'='*50} 生成 50 个等号 + print(f"\n{'='*50}") # '='*50 生成 50 个等号（类似 Java 的 String.repeat(50)） print(f"[思考] 第 {state['iteration']} 轮:") print(thought_text) # 解析 LLM 的输出 - # LLM 输出类似: - # [思考] 我需要计算这个数学题 - # [行动] calculator|2+3*4 - for line in thought_text.split('\n'): # 按行分割 - if '[行动]' in line: # 找到行动行 - # 替换掉 "[行动]"，然后用 "|" 分割 + # .split('\n') 按换行符分割字符串 + # Java: String[] lines = thought_text.split("\n"); + # Python: lines = thought_text.split('\n') + for line in thought_text.split('\n'): + if '[行动]' in line: # 'in' 检查子字符串（类似 Java 的 contains()） + # .replace() 替换字符串 + # .strip() 去前后空格（类似 Java 的 trim()） + # .split('|') 按 | 分割 parts = line.replace('[行动]', '').strip().split('|') - if len(parts) == 2: # 确保有工具名和参数 - state['action'] = parts[0].strip() # 工具名 - state['action_param'] = parts[1].strip() # 参数 - return state # 返回，让路由函数决定下一步 - if '[回答]' in line: # 找到回答行 + if len(parts) == 2: # len() 获取长度（类似 Java 的 .length()） + state['action'] = parts[0].strip() + state['action_param'] = parts[1].strip() + return state + if '[回答]' in line: state['final_answer'] = line.replace('[回答]', '').strip() return state - # 如果没解析到行动或回答，清空 action state['action'] = "" return state def act_node(state: AgentState): - """ - 行动节点 - 执行工具 - - 1. 从 state 中获取工具名和参数 - 2. 在 tools 字典中查找对应的函数 - 3. 调用函数，保存结果到 observation - """ + """行动节点 - 执行工具""" state = state.copy() - action = state.get('action', '') # 工具名，如 "calculator" - param = state.get('action_param', '') # 参数，如 "123*456" + + # .get() 带默认值 + # Java: String action = state.getOrDefault("action", ""); + # Python: action = state.get('action', '') + action = state.get('action', '') + param = state.get('action_param', '') print(f"\n[行动] 执行 {action}({param})") - if action in tools: # 检查工具是否存在 - result = tools[action](param) # 调用对应的函数 + # 检查键是否存在 + # Java: if (tools.containsKey(action)) + # Python: if action in tools + if action in tools: + result = tools[action](param) # 调用函数 state['observation'] = result print(f"[观察] {result}") else: @@ -260,21 +297,16 @@ def act_node(state: AgentState): return state def answer_node(state: AgentState): - """ - 回答节点 - 生成最终答案 - - 两种情况： - 1. LLM 已经在 think_node 中给出了 [回答]，直接用 - 2. 达到最大迭代次数，让 LLM 总结现有信息 - """ + """回答节点 - 生成最终答案""" state = state.copy() - # 情况 1：LLM 已经给出了最终答案 if state.get('final_answer'): print(f"\n[回答] {state['final_answer']}") return state - # 情况 2：需要 LLM 总结 + # 列表推导式（类似 Java 的 Stream） + # Java: String.join("\n", thoughts) + # Python: "\n".join(thoughts) messages = [ {"role": "system", "content": "请根据以下信息给出简洁的最终答案"}, {"role": "user", "content": f"问题: {state['question']}\n\n思考过程:\n" + "\n".join(state.get('thoughts', []))} @@ -288,72 +320,36 @@ def answer_node(state: AgentState): # 6. 路由函数（Router） # ============================================================ """ -路由函数决定图走到哪里。 -它接收当前 State，返回下一步要去的节点名。 +路由函数：接收 State，返回字符串（下一步节点名）。 -为什么需要路由函数？ -- 固定边：永远走同一条路 -- 条件边：根据 State 的内容动态决定走哪条路 -- 智能体需要"判断"能力，所以用条件边 +Java 对比: + // Java: String route(AgentState state) { ... return "act"; } + // Python: def route(state: AgentState) -> str: ... return "act" """ def route(state: AgentState): - """ - 决定下一步去哪里 - - 返回 "act" -> 去执行工具 - 返回 "answer" -> 去生成答案（结束） - """ - # 如果 LLM 已经给出了最终答案，直接结束 + """决定下一步去哪里""" if state.get('final_answer'): return "answer" - - # 如果达到最大迭代次数，强制结束（防止无限循环） if state['iteration'] >= state['max_iterations']: return "answer" - - # 如果有行动要执行，去 act 节点 if state.get('action'): return "act" - - # 默认去 answer 节点 return "answer" # ============================================================ # 7. 构建图（Build Graph） # ============================================================ -""" -StateGraph(State) - 创建图，指定 State 类型 -add_node(name, func) - 添加节点 -add_edge(from, to) - 添加固定边 -add_conditional_edges(node, router, mapping) - 添加条件边 -compile() - 编译图，创建可执行的应用 +graph = StateGraph(AgentState) -图的结构： - START -> think -> [条件边] -> act -> think (循环) - | - +-> answer -> END -""" -graph = StateGraph(AgentState) # 创建图 +graph.add_node("think", think_node) +graph.add_node("act", act_node) +graph.add_node("answer", answer_node) -# 添加三个节点 -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", route, {"act": "act", "answer": "answer"}) +graph.add_edge("act", "think") +graph.add_edge("answer", END) -# 添加边 -graph.add_edge(START, "think") # 开始 -> 思考 - -# 条件边：从 think 出发，根据 route 函数的返回值决定走向 -graph.add_conditional_edges( - "think", # 从 think 节点出发 - route, # 用 route 函数做判断 - {"act": "act", "answer": "answer"} # 返回值 -> 节点名的映射 -) - -graph.add_edge("act", "think") # 行动完回到思考（形成循环！） -graph.add_edge("answer", END) # 回答完结束 - -# 编译图 app = graph.compile() # ============================================================ @@ -370,30 +366,30 @@ print(" START -> think -> [有行动?] -> act -> think (循环)") print(" |") print(" +-> [无行动/达到限制] -> answer -> END") -# 命令行参数处理 -# sys.argv 是命令行参数列表，sys.argv[0] 是脚本名 +# 命令行参数 +# Java: public static void main(String[] args) +# Python: sys.argv[0] 是脚本名，sys.argv[1:] 是参数 if len(sys.argv) > 1: - # 有参数：用参数作为问题 questions = [" ".join(sys.argv[1:])] # 把所有参数拼成一个字符串 else: - # 无参数：使用内置测试问题 questions = [ "计算一下 123 * 456", "什么是 LangGraph？", ] -# 遍历每个问题，依次运行 +# for 循环 +# Java: for (String q : questions) { ... } +# Python: for q in questions: for q in questions: print(f"\n{'#'*50}") print(f"问题: {q}") print(f"{'#'*50}") # app.invoke() 运行图 - # 传入初始 State，图会一步步处理，最终返回完整 State result = app.invoke({ "question": q, - "thoughts": [], # 空列表，记录思考历史 - "iteration": 0, # 从 0 开始 + "thoughts": [], + "iteration": 0, "max_iterations": MAX_ITERATIONS, "action": "", "observation": "", @@ -401,7 +397,6 @@ for q in questions: "action_param": "", }) - # 打印最终结果 print(f"\n{'='*50}") print(f"最终答案: {result['final_answer']}") print(f"{'='*50}") \ No newline at end of file