From 90b0da0e66257c4d0cb994ce529bf903bab37bd5 Mon Sep 17 00:00:00 2001 From: Annyfee <2287551746@qq.com> Date: Sun, 23 Nov 2025 16:32:34 +0800 Subject: [PATCH] feat(langgraph): introduce advanced agent patterns in new module --- 09_langgraph_advanced/01_human_approval.py | 118 +++++++++ 09_langgraph_advanced/02_graph_as_a_tool.py | 119 +++++++++ .../03_multi_agent_orchestration.py | 167 +++++++++++++ .../04_supervisor_agent_ops_system.py | 227 ++++++++++++++++++ 09_langgraph_advanced/README.md | 102 ++++++++ README.md | 47 ++-- 6 files changed, 759 insertions(+), 21 deletions(-) create mode 100644 09_langgraph_advanced/01_human_approval.py create mode 100644 09_langgraph_advanced/02_graph_as_a_tool.py create mode 100644 09_langgraph_advanced/03_multi_agent_orchestration.py create mode 100644 09_langgraph_advanced/04_supervisor_agent_ops_system.py create mode 100644 09_langgraph_advanced/README.md diff --git a/09_langgraph_advanced/01_human_approval.py b/09_langgraph_advanced/01_human_approval.py new file mode 100644 index 0000000..d30e207 --- /dev/null +++ b/09_langgraph_advanced/01_human_approval.py @@ -0,0 +1,118 @@ +import os +from config import OPENAI_API_KEY,LANGCHAIN_API_KEY +from langchain_openai import ChatOpenAI +from langchain.tools import tool +from langchain_core.messages import HumanMessage +from langgraph.graph import StateGraph, MessagesState, START, END +from langgraph.prebuilt import ToolNode +from langgraph.checkpoint.memory import MemorySaver + +# LangSmith调试 +os.environ["LANGCHAIN_TRACING_V2"] = "true" # 总开关，决定启用追踪功能 +os.environ["LANGCHAIN_PROJECT"] = "human_approval" # 自定义项目名 +os.environ["LANGCHAIN_API_KEY"] = LANGCHAIN_API_KEY + +# llm配置 +llm = ChatOpenAI( + model="deepseek-chat", + api_key=OPENAI_API_KEY, + base_url="https://api.deepseek.com" +) + + +# 定义一个敏感工具：发送邮件(模拟) +@tool +def send_email(to, content): + """模拟发送邮件""" + return f'邮件已发送至{to},内容为：{content}' + + +# 工具绑定到llm +tools = [send_email] +llm_with_tools = llm.bind_tools(tools) + +# Node函数与Edge节点 +tool_node = ToolNode(tools) + + +def call_model(state: MessagesState): + response = llm_with_tools.invoke(state['messages']) + return {"messages":[response]} + + +def should_continue(state: MessagesState): + last_msg = state['messages'][-1] + if hasattr(last_msg, "tool_calls") and last_msg.tool_calls: + return "tools" + return END + + + +# 构建基础ReAct图 +workflow = StateGraph(MessagesState) + +workflow.add_node("agent", call_model) +workflow.add_node("tools", tool_node) +workflow.add_edge(START, "agent") + +workflow.add_conditional_edges( + "agent", + should_continue, + { + "tools": "tools", + END: END + } +) + +workflow.add_edge("tools", "agent") + +app = workflow.compile( + # 在内存里做状态持久化 + checkpointer=MemorySaver(), + interrupt_before=["tools"] # 选择要人工审批的节点 -- 负责在哪里停，之后的代码负责停了之后怎么办 +) + + +if __name__ == '__main__': + config = { + "configurable":{"thread_id":"user123"} + } + user_input = "请帮我给 boss@example.com 发一封邮件，内容是：会议推迟到明天下午3点。" + + print("用户输入:",user_input) + print("\nAgent正在思考...\n") + + # 初识输入 + inputs = {"messages":[HumanMessage(content=user_input)]} + + while 1: + # 触发工作流执行，推进到下一个中断点或自然结束 + # inputs注入事件;config确定回话id;"values":完整记录每步结果 + for _ in app.stream(inputs,config,stream_mode="values"): # 流式执行 + pass # 必须迭代生成器，才能实际执行工作流 + + # 获取当前状态 + snapshot = app.get_state(config) + next_tasks = snapshot.next # 返回下一步要执行的节点名列表 + + # 如果没有下一步，说明工作流已结束 + if not next_tasks: + final_msg = snapshot.values['messages'][-1] + print(f'\n最终回复:{final_msg.content}') + break + + # 如果下一步是需要审批的节点 + if "tools" in next_tasks: + last_msg = snapshot.values['messages'][-1] + tool_call = last_msg.tool_calls[0] + print(f'\n⚠️ Agent准备执行操作：') + print(f' 工具名称：{tool_call["name"]}') + print(f' 参数：{tool_call["args"]}') + + approval = input("\n✅ 是否批准执行？(输入 'yes' 继续，其他取消): ").strip().lower() + if approval == "yes": + print('\n 继续执行...') + inputs = None # 表示从断点继续，无新输入 + else: + print("\n❌ 操作已取消，流程终止") + break \ No newline at end of file diff --git a/09_langgraph_advanced/02_graph_as_a_tool.py b/09_langgraph_advanced/02_graph_as_a_tool.py new file mode 100644 index 0000000..7313b6e --- /dev/null +++ b/09_langgraph_advanced/02_graph_as_a_tool.py @@ -0,0 +1,119 @@ +import os +from config import OPENAI_API_KEY,LANGCHAIN_API_KEY +from langchain_openai import ChatOpenAI +from langchain.tools import tool +from langchain_core.messages import HumanMessage, SystemMessage +from langgraph.graph import StateGraph, MessagesState, START, END +from langgraph.prebuilt import ToolNode +from typing import TypedDict + +# LangSmith调试 +os.environ["LANGCHAIN_TRACING_V2"] = "true" # 总开关，决定启用追踪功能 +os.environ["LANGCHAIN_PROJECT"] = "graph_as_tool" # 自定义项目名 +os.environ["LANGCHAIN_API_KEY"] = LANGCHAIN_API_KEY + +# llm配置 +llm = ChatOpenAI( + model="deepseek-chat", + api_key=OPENAI_API_KEY, + base_url="https://api.deepseek.com" +) + +# 构建子工作流 +# 1.子任务状态 +class RetryState(TypedDict): + query: str + attempt: int + result: str + +# 2.子图逻辑 -- 模拟一个可能失败，需重试的API调用 +def call_unstable_api(state:RetryState): + """模拟偶发性的外部服务，偶发失败""" + attempt = state["attempt"] + if attempt == 1: + # 第一次故意失败 + return {"result":"ERROR：服务暂时不可用","attempt":attempt+1} + else: + # 第二次成功 + return {"result":f"SUCCESS：成功处理请求：{state['query']}","attempt":attempt+1} + +def should_retry(state:RetryState): + if "ERROR" in state["result"] and state["attempt"] <= 2: # 出现报错且重试次数小于2，重连 + return "call_api" + return END + +# 3.构建子图工作流 +retry_workflow = StateGraph(RetryState) +retry_workflow.add_node("call_api",call_unstable_api) +retry_workflow.add_edge(START,"call_api") +retry_workflow.add_conditional_edges( + "call_api", + should_retry, + {"call_api":"call_api",END:END} +) +retry_app = retry_workflow.compile() + +# 4.封装为tool(Graph-as-a-Tool) +@tool +def create_order(query:str) -> str: + """创建新订单，自动重试保障成功率""" + result = retry_app.invoke({"query":query,"attempt":1,"result":""}) + return result["result"] + + +# 5.主graph +tools = [create_order] +llm_with_tools = llm.bind_tools(tools) +tool_node = ToolNode(tools) + +def agent_node(state:MessagesState): + response = llm_with_tools.invoke(state["messages"]) + return {"messages":[response]} + +def should_continue(state:MessagesState): + last_msg = state["messages"][-1] + if hasattr(last_msg,"tool_calls") and last_msg.tool_calls: + return "tools" + return END + +# 构建主工作流 +workflow = StateGraph(MessagesState) +workflow.add_node("agent",agent_node) +workflow.add_node("tools",tool_node) +workflow.add_edge(START,"agent") +workflow.add_conditional_edges( + "agent", + should_continue, + { + "tools": "tools", + END: END + } +) +workflow.add_edge("tools","agent") + +app = workflow.compile() + + +# 运行 +if __name__ == '__main__': + user_input = "请创建一个新订单：购买三本书" + print('用户输入:',user_input) + + inputs = {"messages":[ + SystemMessage(content="你是一个任务执行助手。当用户提出任何需要处理、操作或执行的请求时，必须调用 create_order 工具来完成，不要自行回答细节"), + HumanMessage(content=user_input) + ]} + result = app.invoke(inputs) + + tool_result = None + # 在主工作流的消息历史中，查找最近的工具执行结果 + for msg in reversed(result["messages"]): + if msg.type == "tool": # 找到ToolMessage类型消息 + tool_result = msg.content + break + if tool_result: + print(f"\n✅ 直接获取子图返回值:\n{tool_result}") + else: + print("\n❌ 未执行任何工具") + final_reply = result["messages"][-1] + print(f'\n最终回复:\n{final_reply}') \ No newline at end of file diff --git a/09_langgraph_advanced/03_multi_agent_orchestration.py b/09_langgraph_advanced/03_multi_agent_orchestration.py new file mode 100644 index 0000000..ff5e23a --- /dev/null +++ b/09_langgraph_advanced/03_multi_agent_orchestration.py @@ -0,0 +1,167 @@ +import os +from config import OPENAI_API_KEY,LANGCHAIN_API_KEY +from langchain_openai import ChatOpenAI +from langchain.tools import tool +from langchain_core.messages import HumanMessage, SystemMessage +from langgraph.graph import StateGraph, MessagesState, START, END +from langgraph.prebuilt import ToolNode +from typing import TypedDict # 定义数据类型 +from typing import Annotated # 注释说明细节 +from langgraph.graph.message import add_messages + +# LangSmith调试 +os.environ["LANGCHAIN_TRACING_V2"] = "true" # 总开关，决定启用追踪功能 +os.environ["LANGCHAIN_PROJECT"] = "multi_agent_orchestration" # 自定义项目名 +os.environ["LANGCHAIN_API_KEY"] = LANGCHAIN_API_KEY + +# llm配置 +llm = ChatOpenAI( + model="deepseek-chat", + api_key=OPENAI_API_KEY, + base_url="https://api.deepseek.com" +) + +# 模拟工具 +@tool +def search_internal_docs(query:str): + """搜索公司内部文档获取政策信息""" + return "根据公司手册，年假为15天" + +@tool +def search_web(query:str): + """通过搜索引擎获取最新的公开信息""" + return "据TechCrunch报道，LangGraph 0.6已支持持久化记忆" + +@tool +def generate_code(requirement:str): + """根据需求生成可运行的Python代码""" + return "python\nprint('Hello from Code writer!')" + + +# 共享状态定义 +class AgentState(TypedDict): + messages: Annotated[list,add_messages] # 自动累积对话历史 + next_speaker: str + +# 专家节点 +def rag_expert(state:AgentState): + prompt = "你是公司知识库专家，只基于内部文档回答问题。回答应简洁明了，直接给出最终结论。请在回答的最后一行加上：(任务已完成)" + messages = [SystemMessage(content=prompt)]+state['messages'] + tools = [search_internal_docs] + response = llm.bind_tools(tools).invoke(messages) + return {'messages':[response]} + +def web_research(state:AgentState): + prompt = "你是互联网研究员，擅长用搜索引擎获取最新公开信息。回答应简洁明了，直接给出最终结论。请在回答的最后一行加上：(任务已完成)" + messages = [SystemMessage(content=prompt)]+state['messages'] + tools = [search_web] + response = llm.bind_tools(tools).invoke(messages) + return {'messages':[response]} + +def code_writer(state:AgentState): + prompt = "你是python工程师，只生成可运行代码，不解释。回答应简洁明了，直接给出最终结论。请在回答的最后一行加上：(任务已完成)" + messages = [SystemMessage(content=prompt)]+state['messages'] + tools = [generate_code] + response = llm.bind_tools(tools).invoke(messages) + return {'messages':[response]} + +# 总控节点 +def supervisor(state:AgentState): + supervisor_prompt = """ + 你是一个任务协调员。你的目标是管理专家来解决用户的问题。 + + 当前对话需要以下专家参与： + - rag_expert：涉及公司政策、内部流程 + - web_research：涉及外部新闻、公开数据 + - code_writer：需要生成代码 + + 【决策逻辑】 + 1. **检查历史记录**：先看上一个回复是否已经完整回答了用户的初始问题。 + 2. **如果已经回答完毕**：必须输出 'FINISH'。 + 3. **如果尚未回答或需要补充**：根据当前缺少的步骤，选择下一个最合适的专家。 + + 请只输出专家名字或 'FINISH'，不要输出任何其他解释。 + """ + messages = [SystemMessage(content=supervisor_prompt)]+state['messages'] + response = llm.invoke(messages) + next_speaker = response.content.strip() + return {"next_speaker":next_speaker} + +# 路由函数定义 +def route_supervisor(state:AgentState): + if state["next_speaker"]=="FINISH": + return END + return state["next_speaker"] + +def should_continue(state:AgentState): + last_msg = state["messages"][-1] + if hasattr(last_msg,"tool_calls") and last_msg.tool_calls: + return "tools" + return "supervisor" + +def route_after_tool(state:AgentState): + # 工具执行完后，通过next_speaker知道是谁调用的，路由回去 + return state["next_speaker"] + + +# 添加工具节点 +tools = [search_internal_docs,search_web,generate_code] +tool_node = ToolNode(tools) + + +# 构建协作图 +workflow = StateGraph(AgentState) + +# 1. 添加节点 +workflow.add_node("supervisor",supervisor) +workflow.add_node("rag_expert",rag_expert) +workflow.add_node("web_research",web_research) +workflow.add_node("code_writer",code_writer) +workflow.add_node("tools",tool_node) + + +# 2. 总控回路 +workflow.add_edge(START,"supervisor") +workflow.add_conditional_edges("supervisor",route_supervisor) + +# 3. 专家节点的ReAct循环 +for member in ["rag_expert","web_research","code_writer"]: # 为每个专家添加条件边：决定是去执行工具还是回总控 + workflow.add_conditional_edges( + member, + should_continue, + {"tools":"tools","supervisor":"supervisor"} + ) + +# 4.工具节点闭环 +workflow.add_conditional_edges( # 工具执行完，根据next_speaker路由回原来的专家 + "tools", + route_after_tool +) + +app = workflow.compile() + + +# 测试运行 +if __name__ == '__main__': + user_input = "公司年假多少天" + print("用户提问:",user_input) + print('\n开始多智能体协作...\n') + + inputs = {"messages":[HumanMessage(content=user_input)]} + # app是编译好的图，stream()会让图开始运转，并返回一个生成器 + # 图每执行完一个节点，就会产出一个step字典 + for step in app.stream(inputs): + # 因为step是个字典，所以需要拆包拿到 节点名(Node) 与 输出内容(output) + for node,output in step.items(): + # 有工具/专家回复 + if "messages" in output: + msg = output["messages"][-1] + if hasattr(msg,"tool_calls") and msg.tool_calls: + call = msg.tool_calls[0] + print(f"【{node}】调用工具 {call['name']}({call['args']})") + else: + print(f"【{node}】回复：{msg.content}") + # supervisor刚做完决策，确定下个发言人 + elif "next_speaker" in output: + speaker = output["next_speaker"] + print(f"【Supervisor】指定下一位发言人：{speaker}") \ No newline at end of file diff --git a/09_langgraph_advanced/04_supervisor_agent_ops_system.py b/09_langgraph_advanced/04_supervisor_agent_ops_system.py new file mode 100644 index 0000000..8fde71f --- /dev/null +++ b/09_langgraph_advanced/04_supervisor_agent_ops_system.py @@ -0,0 +1,227 @@ +import os +import time +from config import OPENAI_API_KEY,LANGCHAIN_API_KEY +from langchain_openai import ChatOpenAI +from langchain.tools import tool +from langchain_core.messages import HumanMessage,SystemMessage +from langgraph.graph import StateGraph,MessagesState,START,END +from langgraph.prebuilt import ToolNode +from langgraph.checkpoint.memory import MemorySaver +from typing import TypedDict,Annotated +from langgraph.graph.message import add_messages + +# LangSmith调试 +os.environ["LANGCHAIN_TRACING_V2"] = "true" # 总开关，决定启用追踪功能 +os.environ["LANGCHAIN_PROJECT"] = "supervisor_agent_ops_system" # 自定义项目名 +os.environ["LANGCHAIN_API_KEY"] = LANGCHAIN_API_KEY + +llm = ChatOpenAI( + model="deepseek-chat", + api_key=OPENAI_API_KEY, + base_url="https://api.deepseek.com" +) + +# === 一、Graph-as-a-Tool === +# === 模拟一个不稳定的SSH日志查询过程 === + +class SSHState(TypedDict): + target_ip: str + attempt: int + logs: str + +def connect_ssh(state:SSHState): + """模拟SSH连接，第一次连接必定超时""" + print(f' [子图]正在尝试连接服务器{state["target_ip"]}(第{state["attempt"]}次)') + if state['attempt'] == 1: + return {'logs':'ERROR:Connection Timed Out','attempt':state['attempt']+1} + return {'logs':'CONNECTED','attempt':state['attempt']+1} + +def grep_system_logs(state:SSHState): + """连接成功后读取日志""" + if state["logs"] == "CONNECTED": + # 打印查到的结果 + return {'logs':f'SUCCESS: Retrieved logs from {state['target_ip']}:[ERROR: OutOfMemory at line 4032]'} + return {'logs':state['logs']} # 保持错误状态 + +def ssh_routing(state:SSHState): + """路由逻辑：如果连接失败且尝试次数少于3，重试""" + if "ERROR" in state['logs'] and state["attempt"] <= 2: + return "connect" + return "grep" + +# 构建子图 +ssh_workflow = StateGraph(SSHState) +ssh_workflow.add_node("connect",connect_ssh) +ssh_workflow.add_node("grep",grep_system_logs) + +ssh_workflow.add_edge(START,"connect") +ssh_workflow.add_conditional_edges("connect",ssh_routing,{"connect":"connect","grep":"grep"}) +ssh_workflow.add_edge("grep",END) + +ssh_app = ssh_workflow.compile() + +# 将子图封装为工具 +@tool +def analyze_server_logs(ip_address:str): + """使用SSH连接服务器并分析最近的错误日志（内含自动重连机制）""" + result = ssh_app.invoke({"target_ip":ip_address,"attempt":1,"log":""}) + return result['logs'] + + + +# === 二、Human-in-the-Loop === +# === 重启服务，高危操作，需要审批 === + +@tool +def restart_service(service_name:str): + """重启指定的服务器服务""" + return f"服务[{service_name}]已成功重启，系统负载已恢复正常" + + + +# === 三、Multi-Agent 编排 === +# === 总控调度 + 专家分工 === + +# 1. 共享状态 +class AgentState(TypedDict): + messages:Annotated[list,add_messages] + next_speaker:str + +# 2. 专家节点 +def log_expert(state:AgentState): + prompt = "你是日志分析专家，使用工具分析服务器日志，找出报错原因。回答需简洁。" + messages = [SystemMessage(content=prompt)] + state['messages'] + # 绑定子图工具 + tools = [analyze_server_logs] + response = llm.bind_tools(tools).invoke(messages) + return {"messages":[response]} + +def ops_expert(state:AgentState): + prompt = "你是运维专家。当收到修复指令时，请立即调用 'restart_service' 工具进行修复，不要输出任何额外的解释文本。" + messages = [SystemMessage(content=prompt)] + state['messages'] + tools = [restart_service] + # 绑定敏感工具 + response = llm.bind_tools(tools).invoke(messages) + return {"messages":[response]} + +# 3. 总控节点（supervisor） +def supervisor(state:AgentState): + prompt = """ + 你是 IT 运维总指挥。 + 专家列表： + - log_expert + - ops_expert + + 决策逻辑： + 1. 未知原因 -> log_expert + 2. 已知原因（如OOM、报错） -> ops_expert + 3. 修复完成 -> FINISH + + 【输出约束】 + 仅输出下一个专家的名字（如 log_expert），不要包含任何其他字符或标点。 + """ + messages = [SystemMessage(content=prompt)] + state['messages'] + response = llm.invoke(messages) + return {"next_speaker":response.content.strip()} + +# 4. 路由逻辑 +def route_supervisor(state:AgentState): + if state['next_speaker'] == "FINISH": + return END + return state['next_speaker'] + +def should_continue(state:AgentState): + last_msg = state['messages'][-1] + if hasattr(last_msg,"tool_calls") and last_msg.tool_calls: + return "tools" + return "supervisor" + +def route_after_tool(state:AgentState): + return state["next_speaker"] + + +# === 四、构建主图与集成 === + +# 工具集合 +all_tools = [analyze_server_logs,restart_service] +tool_node = ToolNode(all_tools) + +# 构建主图 +workflow = StateGraph(AgentState) + +workflow.add_node("supervisor",supervisor) +workflow.add_node("log_expert",log_expert) +workflow.add_node("ops_expert",ops_expert) +workflow.add_node("tools",tool_node) + +workflow.add_edge(START,"supervisor") +workflow.add_conditional_edges("supervisor",route_supervisor) +for member in ["log_expert","ops_expert"]: + workflow.add_conditional_edges( + member, + should_continue, + {"tools":"tools","supervisor":"supervisor"} + ) +workflow.add_conditional_edges("tools",route_after_tool) + +# 编译图:加入记忆与中断机制(Human-in-the-Loop) +# 注: 我们在所有工具执行前都暂停，但在运行时进行逻辑判断 +app = workflow.compile( + checkpointer=MemorySaver(), + interrupt_before=["tools"] +) + +# === 五、运行时逻辑(模拟生产环境的交互) === + +if __name__ == '__main__': + # 模拟一次完整的故障处理流程 + user_input = "服务器 192.168.1.100 报警，响应极慢，请处理。" + config = { + "configurable": {"thread_id": "incident_001"} + } + + print(f'收到报警 : {user_input}') + inputs = {"messages":[HumanMessage(content=user_input)]} + + # 循环执行，直到任务结束 + while 1: + # 1. 执行图直到中断或结束 + for _ in app.stream(inputs,config,stream_mode="values"): + pass + + # 2. 检查当前状态 + snapshot = app.get_state(config) + next_tasks = snapshot.next + + # 没有下一步，任务结束 + if not next_tasks: + print(f" 最终报告:{snapshot.values['messages'][-1].content}") + break + + # 3. 处理中断:判断是哪个工具被调用 + if "tools" in next_tasks: + last_msg = snapshot.values['messages'][-1] + tool_call = last_msg.tool_calls[0] + tool_name = tool_call["name"] + + print(f'\n[系统暂停] 请求调用工具:{tool_name}') + + # 策略A: 自动放行安全工具(Graph-as-a-Tool) + if tool_name == "analyze_server_logs": + print(' -> 这是一个查询类工具，系统自行批准。') + inputs = None # 继续执行 + continue + + # 策略B: 拦截高危工具(Human-in-the-Loop) + elif tool_name == "restart_service": + print(" -> ⚠️ 警告: 这是一个高危操作!") + user_approval = input(" -> 请人工审批 (输入 'yes' 允许重启):") + + if user_approval == "yes": + print(' -> ✅ 审批通过，正在执行...') + inputs = None # 继续执行 + else: + print(' -> ❌️ 审批拒绝，任务终止!') + # 实际系统中，应通过 ToolMessage 反馈人工拒绝，使 LLM 能继续响应； + # 当前demo为简化，直接退出 + break \ No newline at end of file diff --git a/09_langgraph_advanced/README.md b/09_langgraph_advanced/README.md new file mode 100644 index 0000000..f0d0595 --- /dev/null +++ b/09_langgraph_advanced/README.md @@ -0,0 +1,102 @@ +# 🧩 模块说明：LangGraph 进阶篇 - 构建企业级智能 Agent 系统 + +> 📌 核心知识点：Human-in-the-Loop 人工干预｜Graph-as-a-Tool 子图封装｜Multi-Agent 多智能体协作｜Supervisor 总控调度｜复杂工作流编排 + +--- + +### 1. `01_human_approval.py` （人工审批机制） + +实现带有人工干预的智能体工作流，为敏感操作添加审批环节。 + +- ✅ 掌握点： + - 在 LangGraph 中使用 `interrupt_before=["tools"]` 设置中断点 + - 通过 `app.get_state(config)` 获取工作流当前状态 + - 实现用户交互审批流程，控制高危操作的执行 + - 结合 MemorySaver 实现持久化状态管理 + +- 输出： + - 当智能体尝试执行敏感操作（如发送邮件）时，系统会暂停并请求用户确认 + - 用户输入 "yes" 批准执行，输入其他取消操作 + - 完整展示执行过程的日志和最终结果 + +> 💡 此示例适用于需要人工监督的场景，如金融交易、系统配置修改等高风险操作。 + +--- + +### 2. `02_graph_as_a_tool.py` （子图工具化封装） + +演示如何将一个完整的 LangGraph 子工作流封装为单个工具（Graph-as-a-Tool 模式）。 + +- ✅ 掌握点： + - 定义独立的子图状态结构（如 `RetryState`） + - 构建包含重试逻辑的子图工作流 + - 使用 `@tool` 装饰器将子图封装为可调用工具 + - 在主工作流中像使用普通工具一样调用子图 + +- 效果： + - 子图内部实现了失败重试逻辑，对调用者完全透明 + - 主工作流可以无缝集成复杂的子流程，保持代码整洁 + - 模拟实现了不稳定 API 调用的容错处理 + +> 💡 这种模式特别适合封装具有复杂内部逻辑的操作，如网络请求、数据库事务等可能需要重试或错误处理的场景。 + +--- + +### 3. `03_multi_agent_orchestration.py` （多智能体协作系统） + +构建由总控节点协调多个专家智能体的协作系统，实现专业化分工。 + +- ✅ 掌握点： + - 定义共享的多智能体状态结构（`AgentState`） + - 创建不同功能的专家节点（RAG专家、网络研究员、代码编写者） + - 实现 Supervisor 总控节点，负责任务分配和流程协调 + - 为每个专家配置专属工具和系统指令 + - 设计复杂的条件路由逻辑，支持多节点协作 + +- 效果： + - 系统会根据用户问题自动选择最合适的专家处理 + - 专家可以调用专业工具获取信息 + - 总控节点监控对话进度，决定是否完成任务或切换专家 + - 完整展示多智能体交互过程和决策路径 + +> 💡 此架构适用于需要多领域专业知识协作解决的复杂问题，如技术咨询、客户服务等场景。 + +--- + +### 4. `04_supervisor_agent_ops_system.py` （企业级运维智能体系统） + +整合前述所有高级特性，构建一个完整的企业级 IT 运维智能体系统。 + +- ✅ 掌握点： + - **Graph-as-a-Tool**：封装 SSH 日志分析子图，处理连接重试逻辑 + - **Human-in-the-Loop**：对重启服务等高危操作实施人工审批 + - **Multi-Agent 协作**：实现日志专家和运维专家的专业分工 + - **Supervisor 调度**：根据问题类型和解决进度智能分配任务 + - **状态持久化**：使用 MemorySaver 维护跨轮次的会话状态 + +- 效果： + - 完整模拟 IT 故障处理流程：报警 → 日志分析 → 故障定位 → 人工审批 → 执行修复 + - 自动化程度高，但保留人工干预的安全机制 + - 各组件松耦合，易于扩展和维护 + - 支持 LangSmith 全流程追踪和调试 + +> 💡 这是一个接近生产环境的综合示例，展示了 LangGraph 构建复杂智能体系统的强大能力。 + +--- + +### 🔔 全局注意事项 + +- **学习路径建议**： + `01`（人工干预） → `02`（子图封装） → `03`（多智能体基础） → `04`（企业级综合应用） +- 所有 `.py` 文件均依赖根目录 `.env` 中的 `OPENAI_API_KEY` 和 `LANGCHAIN_API_KEY` +- LangSmith 追踪功能已默认开启，可在 [LangSmith UI](https://smith.langchain.com) 实时查看执行流程 +- `04_supervisor_agent_ops_system.py` 是一个整合性示例，建议在理解前三个文件的基础上学习 + +--- + +### 💡 **建议** +- 尝试扩展 `04_supervisor_agent_ops_system.py`，添加更多专业角色（如安全专家、性能分析师） +- 实验不同的 Supervisor 决策逻辑，优化任务分配策略 +- 将真实的运维工具或 API 集成到系统中，构建实用的运维助手 +- 探索在生产环境中部署此类系统的最佳实践，如性能优化、错误处理和监控 +- 考虑添加更复杂的工作流模式，如并行执行、超时处理等高级特性 \ No newline at end of file diff --git a/README.md b/README.md index 8c79a4b..d9f2723 100644 --- a/README.md +++ b/README.md @@ -28,32 +28,32 @@ ## 🧭 模块总览（16 模块） > 💡 每个模块对应独立目录（含代码、说明与示例），可独立运行与学习。 -> 🔥 已更新至 08 LangGraph 基础篇，持续更新中！ +> 🔥 已更新至 09 LangGraph 进阶篇，持续更新中！ -| 章节 | 模块 | 博客 | 核心关键词 | 难度 | -|:-----------:|:--------------------------------------------------------------------------------------------------------|:-------------------------------------------------------------------:|:----------------------------------------------------|:-----:| -| 🏗️ **基础篇** | [01 Agent 入门 & 环境搭建](https://github.com/Annyfee/agent-craft/tree/main/01_agent_introduction) | [🏠](https://blog.csdn.net/2401_87328929/article/details/153729555) | OpenAI API | ⭐ | -| | [02 LLM 基础调用](https://github.com/Annyfee/agent-craft/tree/main/02_llm_fundamentals) | [🏠](https://blog.csdn.net/2401_87328929/article/details/153735431) | LLM API 调用 · prompt · 上下文记忆 | ⭐ | -| | [03 Function Calling 与工具调用](https://github.com/Annyfee/agent-craft/tree/main/03_function_calling_tools) | [🏠](https://blog.csdn.net/2401_87328929/article/details/153866573) | Function Call · 工具函数封装 | ⭐⭐ | -| ⚙️ **框架篇** | [04 LangChain 基础篇](https://github.com/Annyfee/agent-craft/tree/main/04_langchain_basics) | [🏠](https://blog.csdn.net/2401_87328929/article/details/153978186) | LLM · Prompt · Chain · Memory | ⭐⭐ | -| | [05 LangChain 进阶篇](https://github.com/Annyfee/agent-craft/tree/main/05_langchain_advanced) | [🏠](https://blog.csdn.net/2401_87328929/article/details/154064397) | Agents · 缓存 · 流式输出 | ⭐⭐⭐ | -| | [06 RAG 基础篇](https://github.com/Annyfee/agent-craft/tree/main/06_rag_basics) | [🏠](https://blog.csdn.net/2401_87328929/article/details/154230067) | RAG概念 · Split · Embedding · FAISS · RAG 链 | ⭐⭐ | -| | [07 RAG 进阶篇](https://github.com/Annyfee/agent-craft/tree/main/07_rag_advanced) | [🏠](https://blog.csdn.net/2401_87328929/article/details/154408744) | Chroma · Reranker · RAG工具化 · 集成 | ⭐⭐⭐ | -| | [08 LangGraph 基础篇](https://github.com/Annyfee/agent-craft/tree/main/08_langgraph_basics) | [🏠](https://blog.csdn.net/2401_87328929/article/details/154576812) | Langgraph三要素 · LangSmith · ReAct · 提示词安全注入 | ⭐⭐⭐ | -| | 09 LangGraph 进阶篇 | 🚧撰写中 | `Multi-Agent` · `Human-in-the-Loop` · “总控-员工”架构 | ⭐⭐⭐⭐ | -| 🧠 **智能篇** | 10 MCP 基础 (Client) | 🚧 | `MCP` 协议 · `CherryStudio` · 消费公共 MCP | ⭐⭐⭐ | -| | 11 MCP 进阶 (Server) | 🚧 | `Streamable HTTP` · `src layout` · 构建私有 MCP | ⭐⭐⭐⭐ | -| | 12 Swarm & Agents SDK | 🚧 | `Swarm` · `Handoff` · “航空公司客服”项目 | ⭐⭐⭐⭐ | -| 🏁 **实战篇** | 13 Streamlit 快速入门 | 🚧 | `st.chat_input` · `st.session_state` · 封装 RAG Agent | ⭐⭐⭐ | -| | 14 综合实战项目 | 🚧 | LangGraph + RAG + MCP + Streamlit + Vercel | ⭐⭐⭐⭐⭐ | -| 🚀 **工程篇** | 15 部署与本地化 | 🚧 | `Ollama` · `LM Studio` · `LangServe` | ⭐⭐⭐ | -| | **16 项目打包与发布** | 🚧 | `pyproject.toml` · `pip build` · `setuptools` · 发布 | ⭐⭐⭐⭐ | +| 章节 | 模块 | 博客 | 核心关键词 | 难度 | +|:-----------:|:--------------------------------------------------------------------------------------------------------|:-------------------------------------------------------------------:|:-------------------------------------------------------------|:-----:| +| 🏗️ **基础篇** | [01 Agent 入门 & 环境搭建](https://github.com/Annyfee/agent-craft/tree/main/01_agent_introduction) | [🏠](https://blog.csdn.net/2401_87328929/article/details/153729555) | OpenAI API | ⭐ | +| | [02 LLM 基础调用](https://github.com/Annyfee/agent-craft/tree/main/02_llm_fundamentals) | [🏠](https://blog.csdn.net/2401_87328929/article/details/153735431) | LLM API 调用 · prompt · 上下文记忆 | ⭐ | +| | [03 Function Calling 与工具调用](https://github.com/Annyfee/agent-craft/tree/main/03_function_calling_tools) | [🏠](https://blog.csdn.net/2401_87328929/article/details/153866573) | Function Call · 工具函数封装 | ⭐⭐ | +| ⚙️ **框架篇** | [04 LangChain 基础篇](https://github.com/Annyfee/agent-craft/tree/main/04_langchain_basics) | [🏠](https://blog.csdn.net/2401_87328929/article/details/153978186) | LLM · Prompt · Chain · Memory | ⭐⭐ | +| | [05 LangChain 进阶篇](https://github.com/Annyfee/agent-craft/tree/main/05_langchain_advanced) | [🏠](https://blog.csdn.net/2401_87328929/article/details/154064397) | Agents · 缓存 · 流式输出 | ⭐⭐⭐ | +| | [06 RAG 基础篇](https://github.com/Annyfee/agent-craft/tree/main/06_rag_basics) | [🏠](https://blog.csdn.net/2401_87328929/article/details/154230067) | RAG概念 · Split · Embedding · FAISS · RAG 链 | ⭐⭐ | +| | [07 RAG 进阶篇](https://github.com/Annyfee/agent-craft/tree/main/07_rag_advanced) | [🏠](https://blog.csdn.net/2401_87328929/article/details/154408744) | Chroma · Reranker · RAG工具化 · 集成 | ⭐⭐⭐ | +| | [08 LangGraph 基础篇](https://github.com/Annyfee/agent-craft/tree/main/08_langgraph_basics) | [🏠](https://blog.csdn.net/2401_87328929/article/details/154576812) | Langgraph三要素 · LangSmith · ReAct · 提示词安全注入 | ⭐⭐⭐ | +| | [09 LangGraph 进阶篇]() | [🏠]() | Human-in-the-Loop · Graph-as-a-Tool · Multi-Agent 多智能体编排 | ⭐⭐⭐⭐ | +| 🧠 **智能篇** | 10 MCP 基础 (Client) | 🚧撰写中 | `MCP` 协议 · `CherryStudio` · 消费公共 MCP | ⭐⭐⭐ | +| | 11 MCP 进阶 (Server) | 🚧 | `Streamable HTTP` · `src layout` · 构建私有 MCP | ⭐⭐⭐⭐ | +| | 12 Swarm & Agents SDK | 🚧 | `Swarm` · `Handoff` · “航空公司客服”项目 | ⭐⭐⭐⭐ | +| 🏁 **实战篇** | 13 Streamlit 快速入门 | 🚧 | `st.chat_input` · `st.session_state` · 封装 RAG Agent | ⭐⭐⭐ | +| | 14 综合实战项目 | 🚧 | LangGraph + RAG + MCP + Streamlit + Vercel | ⭐⭐⭐⭐⭐ | +| 🚀 **工程篇** | 15 部署与本地化 | 🚧 | `Ollama` · `LM Studio` · `LangServe` | ⭐⭐⭐ | +| | **16 项目打包与发布** | 🚧 | `pyproject.toml` · `pip build` · `setuptools` · 发布 | ⭐⭐⭐⭐ | --- ## 🧩 当前学习建议（适用于已完成模块） -目前已开放模块：**01 ~ 08**。 +目前已开放模块：**01 ~ 09**。 建议按顺序学习，每一章都有完整代码示例与注释。 ### ✅ 模块 01 — Agent 入门 & 环境搭建 @@ -96,7 +96,12 @@ - **目标**：懂得Langgraph的三要素，会使用LangSmith调试当前agent，并能自己写一个简单的Langgraph流程。 - **内容**：Langgraph核心三要素｜ReAct循环白盒化实现 | LangSmith调试追踪 | 持久化记忆 | 提示词安全注入 -> 📌 后续模块将陆续开放（MCP、多智能体、Streamlit等） +### ✅ 模块 09 — LangGraph 进阶篇 + +- **目标**：学会人工审批机制的添加，子图工具化的封装与多智能体流程编排。最终可以融合这三个知识点做一个复杂Agent出来。 +- **内容**：Human-in-the-Loop（人工干预）｜ Graph-as-a-Tool（图即工具）| Multi-Agent 多智能体编排 + +> 📌 后续模块将陆续开放（MCP、Streamlit、Swarm等） ---