甜点网站建设的功能及意义2017商会网站建设方案

甜点网站建设的功能及意义,2017商会网站建设方案,网站制作咨询公司,网站数据库连接出错本文详细解析了LangGraph框架中的Supervisor#xff08;代理主管#xff09;机制#xff0c;通过微软Magentic-One系统引出多智能体协作架构。文章介绍了Supervisor分工明确中央协调动态路由的核心思想#xff0c;提供了完整代码实现和数据库分析案例#xff…本文详细解析了LangGraph框架中的Supervisor代理主管机制通过微软Magentic-One系统引出多智能体协作架构。文章介绍了Supervisor分工明确中央协调动态路由的核心思想提供了完整代码实现和数据库分析案例展示了如何构建可扩展、自适应的多智能体系统。同时指出Supervisor可能出现的循环决策问题并总结了其在结构化任务调度、可扩展代理体系和智能决策方面的价值为构建高效协作的多智能体系统提供了实用指导。在多智能体系统Multi-Agent System的发展历程中LangGraph 正逐渐成为最具代表性的编排框架之一。它不仅能让多个智能体协同工作还能通过图结构管理复杂的工作流。而在众多特性中最值得关注的莫过于它所引入的“Supervisor代理主管”机制——一个能够像项目经理一样统筹规划、任务分派、结果整合的核心控制单元。这篇文章我们将从微软的最新多智能体系统 Magentic-One 入手深入理解 Supervisor 架构的思想来源与实现逻辑最后带你一步步在 LangGraph 中构建一个真正可运行的多智能体系统。一、从 Magentic-One 说起多智能体系统的协作样本微软在 2024 年推出的 Magentic-One可以说是当前多智能体系统的“教科书级”范例。它允许多个 AI 代理协同工作每个代理都是某一领域的专家彼此之间通过中心控制器Orchestrator进行通信与协调。举个例子在一个软件开发场景中WebSurfer代理负责网页搜索FileSurfer负责文件管理Coder负责编写代码ComputerTerminal负责执行程序而Orchestrator则作为“大脑”调度上述所有代理。这种设计让整个系统具备了像人类团队一样的协作模式一个主管带领多个执行者实时分工、汇报、反馈、调整。如下图所示Magentic-One 的架构逻辑清晰分层在实际任务中比如提取图片中的 Python 代码、运行后生成 C 源码再执行得到计算结果系统的任务流是这样运转的FileSurfer提取 Python 代码Coder分析代码逻辑ComputerTerminal执行 PythonWebSurfer打开 URL 获取 C 源码Coder再次分析最终由ComputerTerminal执行并返回结果。整个过程由 Orchestrator 全程监督调度。这套体系的关键优势在于各代理功能独立但可协作流程具备弹性和可恢复性具备“动态自适应”的能力能根据环境变化实时调整。这正是 LangGraph 的 Supervisor 架构 所要实现的目标。二、LangGraph的Supervisor架构是什么它的核心思想就是分工明确 中央协调 动态路由。LangGraph 的 Supervisor 就是把这种思想“图化”把调度逻辑用状态图(StateGraph)表达出来便于可视化、调试与扩展。它作为中央调度中心负责协调多个子代理Worker Agents的工作流。简单说它就是图中的一个特殊节点充当多代理工作流中的中央控制器。整个流程是这样的用户输入 → Supervisor接收 → 判断该让谁处理 → 指派任务给对应代理 → 代理执行完后向Supervisor汇报 → Supervisor决定是继续还是结束。这种架构有个巨大的优势灵活性。随着需求变化或新代理的加入系统可以动态调整工作流程。特别是在客户服务这类动态环境中查询和需求不断变化这种架构就显得特别有用。三、如何用代码实现我们来看一个实际的例子。假设要搭建一个包含三个功能的系统Chat代理处理常规对话Coder代理处理代码生成和执行SQLer代理处理数据库相关操作首先要定义系统的状态。在LangGraph中通过StateGraph来实现class AgentState(MessagesState): next: str然后定义Supervisor节点的行为。这个节点需要读取当前的消息分析用户需求然后决定下一步该调用哪个代理。关键技术是利用结构化输出强制LLM返回特定格式class Router(TypedDict): next: Literal[chat, coder, sqler, FINISH]next字段就是Supervisor的决策结果。然后分别定义三个子代理节点。这里有个很重要的细节——每个代理在返回结果时都要标上自己的名字。这样Supervisor才能知道这条消息是谁产生的def chat(state: AgentState): messages state[messages][-1] model_response llm.invoke(messages.content) final_response [HumanMessage(contentmodel_response.content, namechat)] return {messages: final_response}最后把所有节点和边连接起来builder StateGraph(AgentState) builder.add_node(supervisor, supervisor) builder.add_node(chat, chat) builder.add_node(coder, coder) builder.add_node(sqler, sqler) # 所有代理完成任务后都向主管汇报 for member in members: builder.add_edge(member, supervisor) # 根据Supervisor的决策进行路由 builder.add_conditional_edges(supervisor, lambda state: state[next]) # 从开始就连接到Supervisor builder.add_edge(START, supervisor) graph builder.compile()这样就搭好了基本框架。测试时问它帮我生成一个二分查找的Python代码系统会自动判断该用Coder代理问我想查询数据库中的数据它就会调用SQLer代理。四、实战案例构建具备数据库与分析能力的多代理系统我们可以把这个思路用到一个真实场景搭建一个包含数据库管理员和数据分析师的多代理系统。数据库管理员db_agent负责对接数据库提取数据。数据分析师code_agent拿到数据后用Python进行分析生成可视化图表。Supervisor在其中扮演协调者的角色判断用户需求然后分别指派任务。用户说帮我生成前10名销售记录的柱状图流程就是Supervisor分析这需要数据库代理和代码代理先让db_agent查询数据库拿出前10名的销售记录再让code_agent根据这些数据生成柱状图最后Supervisor收集结果返回给用户整个过程对用户来说是无感知的但背后的协调工作很复杂。下面是一份完整的 Python 脚本基于 SQLite包含三个子代理chat、sqler、coder和一个 Supervisor。# 文件名langgraph_supervisor.py #LangGraph Supervisor 多代理示例SQLite 后端 Python 执行工具 import os import getpass import random from typing import Literal from typing_extensions import TypedDict # 数据库 from sqlalchemy import create_engine, Column, Integer, String, Float from sqlalchemy.orm import sessionmaker, declarative_base # 生成假数据 from faker import Faker # Pydantic from pydantic import BaseModel # LLM/工具和 LangGraph按本地版本调整导入 try: from langchain_openai import ChatOpenAI except Exception: from langchain import OpenAI as ChatOpenAI # 兼容方案 try: from langchain_core.tools import tool from langchain_core.messages import HumanMessage except Exception: from langchain.tools import tool from langchain.schema import HumanMessage try: from langgraph.graph import StateGraph, MessagesState, START, END from langgraph.prebuilt import create_react_agent except Exception as e: raise RuntimeError(请确认已安装并可导入 langgraph。按需调整导入路径。) # 可选的 PythonREPL try: from langchain_experimental.utilities import PythonREPL except Exception: PythonREPL None # ------------------------- # 1. 配置 LLM # ------------------------- if not os.environ.get(OPENAI_API_KEY): os.environ[OPENAI_API_KEY] getpass.getpass(请输入你的 OPENAI_API_KEY: ) # 这里示例用 gpt-4o-mini如无此模型请替换为本地或小模型 llm ChatOpenAI(modelgpt-4o-mini) # ------------------------- # 2. 本地 SQLite 数据库与 ORM # ------------------------- Base declarative_base() class SalesData(Base): __tablename__ sales_data sales_id Column(Integer, primary_keyTrue, autoincrementTrue) product_id Column(Integer) employee_id Column(Integer) customer_id Column(Integer) sale_date Column(String(50)) quantity Column(Integer) amount Column(Float) discount Column(Float) DB_FILE sales_demo_cn.sqlite DB_URI fsqlite:///{DB_FILE} engine create_engine(DB_URI, echoFalse) Base.metadata.create_all(engine) Session sessionmaker(bindengine) def seed_data_if_empty(n100): session Session() if session.query(SalesData).count() 0: session.close() return fake Faker() for _ in range(n): s SalesData( product_idrandom.randint(1, 10), employee_idrandom.randint(1, 8), customer_idrandom.randint(1, 40), sale_datefake.date_between(start_date-120d, end_datetoday).isoformat(), quantityrandom.randint(1, 20), amountround(random.uniform(10, 1200), 2), discountround(random.uniform(0, 0.3), 2) ) session.add(s) session.commit() session.close() print(f已初始化并插入 {n} 条示例数据到 {DB_FILE}) seed_data_if_empty(120) # ------------------------- # 3. 定义工具 # ------------------------- class AddSaleSchema(BaseModel): product_id: int employee_id: int customer_id: int sale_date: str quantity: int amount: float discount: float class DeleteSaleSchema(BaseModel): sales_id: int class UpdateSaleSchema(BaseModel): sales_id: int quantity: int amount: float class QuerySalesSchema(BaseModel): sales_id: int tool(args_schemaAddSaleSchema) def add_sale(product_id, employee_id, customer_id, sale_date, quantity, amount, discount): 添加销售记录 session Session() try: rec SalesData( product_idproduct_id, employee_idemployee_id, customer_idcustomer_id, sale_datesale_date, quantityquantity, amountamount, discountdiscount ) session.add(rec) session.commit() return {messages: [添加成功。], sales_id: rec.sales_id} except Exception as e: return {messages: [f添加失败错误{e}]} finally: session.close() tool(args_schemaDeleteSaleSchema) def delete_sale(sales_id): 删除销售记录 session Session() try: rec session.query(SalesData).filter(SalesData.sales_id sales_id).first() if not rec: return {messages: [f未找到 sales_id{sales_id} 的记录。]} session.delete(rec) session.commit() return {messages: [f删除成功{sales_id}]} except Exception as e: return {messages: [f删除失败错误{e}]} finally: session.close() tool(args_schemaUpdateSaleSchema) def update_sale(sales_id, quantity, amount): 更新销售记录 session Session() try: rec session.query(SalesData).filter(SalesData.sales_id sales_id).first() if not rec: return {messages: [f未找到 sales_id{sales_id} 的记录。]} rec.quantity quantity rec.amount amount session.commit() return {messages: [f更新成功{sales_id}]} except Exception as e: return {messages: [f更新失败错误{e}]} finally: session.close() tool(args_schemaQuerySalesSchema) def query_sales(sales_id): 查询销售记录 session Session() try: rec session.query(SalesData).filter(SalesData.sales_id sales_id).first() if not rec: return {messages: [f未找到 sales_id{sales_id} 的记录。]} return { sales_id: rec.sales_id, product_id: rec.product_id, employee_id: rec.employee_id, customer_id: rec.customer_id, sale_date: rec.sale_date, quantity: rec.quantity, amount: rec.amount, discount: rec.discount } except Exception as e: return {messages: [f查询失败错误{e}]} finally: session.close() # Python REPL用于 code_agent 执行生成的 Python请勿在非受控环境执行任意代码 if PythonREPL is not None: repl PythonREPL() tool def python_repl(code: str): 执行 python 代码并返回 stdout try: result repl.run(code) return f执行成功输出{result} except Exception as e: return f执行失败错误{repr(e)} else: tool def python_repl(code: str): 简易执行器仅用于本地演示禁止危险模块 banned [os, sys, subprocess, socket, shutil] if any(b in code for b in banned): return 拒绝执行包含危险模块的代码。 local_vars {} try: exec(code, {__builtins__: __builtins__}, local_vars) return f执行结束本地变量{list(local_vars.keys())} except Exception as e: return f执行失败错误{repr(e)} # ------------------------- # 4. 创建 ReAct 风格的 Agents # ------------------------- db_agent create_react_agent( llm, tools[add_sale, delete_sale, update_sale, query_sales], state_modifier你是数据库代理。接收用户请求并调用数据库工具增删改查。返回结果需清晰、结构化必要时返回 sales_id。 ) code_agent create_react_agent( llm, tools[python_repl], state_modifier你是代码代理。根据输入的数据生成可运行的 Python 代码来分析或绘图并执行后返回结果或文件路径。若生成代码请在最后用 print() 输出关键结果。 ) # ------------------------- # 5. 将 agents 封装为图节点回报时包含 name 字段 # ------------------------- def db_node(state): res db_agent.invoke(state) content res[messages][-1].content if res[messages] else 无返回 # 把 name 设置为 sqler 便于 supervisor 识别 return {messages: [HumanMessage(contentcontent, namesqler)]} def code_node(state): res code_agent.invoke(state) content res[messages][-1].content if res[messages] else 无返回 return {messages: [HumanMessage(contentcontent, namecoder)]} def chat_node(state): # chat 用 LLM 直接回答 msg state[messages][-1] resp llm.invoke(msg.content) return {messages: [HumanMessage(contentresp.content, namechat)]} # ------------------------- # 6. Supervisor # ------------------------- members [chat, coder, sqler] options members [FINISH] class Router(TypedDict): next: Literal[chat, coder, sqler, FINISH] class AgentState(MessagesState): next: str def supervisor(state: AgentState): system_prompt ( 你是一个监督者Supervisor负责在以下代理之间分配任务chat, coder, sqler。\n\n 每个代理的职责\n - chat使用自然语言直接回应用户问题做一般性解释与对话。\n - coder生成并在安全沙箱中执行 Python 代码用于数据分析或绘图返回执行结果或图像路径。\n - sqler执行数据库相关操作增删改查并返回结构化数据或操作状态。\n\n 请根据下列对话上下文判断接下来应该由哪个代理执行任务并以 JSON 格式仅返回一个字段 next值必须是以下之一 f{options}。如果任务已经完成请返回 FINISH。\n\n 注意输出必须是严格的 JSON例如{\next\:\sqler\}。不要返回其他任何多余文本。 ) messages [{role: system, content: system_prompt}] state[messages] try: # 优先使用结构化输出能力若 LLM wrapper 支持 response llm.with_structured_output(Router).invoke(messages) next_ response[next] except Exception: # 作为后备方案解析模型的文本回答寻找关键字 raw llm.invoke(messages).content nxt None for m in options: if m in raw: nxt m break next_ nxt or FINISH if next_ FINISH: next_ END return {next: next_} # ------------------------- # 7. 构建 StateGraph # ------------------------- builder StateGraph(AgentState) builder.add_node(supervisor, supervisor) builder.add_node(chat, chat_node) builder.add_node(coder, code_node) builder.add_node(sqler, db_node) # 子代理执行后回到 supervisor for mem in members: builder.add_edge(mem, supervisor) # conditional edge由 state[next] 指定 builder.add_conditional_edges(supervisor, lambda state: state[next]) builder.add_edge(START, supervisor) graph builder.compile() # 显示图可选 try: from IPython.display import Image, display display(Image(graph.get_graph(xrayTrue).draw_mermaid_png())) except Exception: pass # ------------------------- # 8. 测试与演示 # ------------------------- def demo_queries(): qs [ 你好请用简单介绍你自己。, 请帮我查询 sales_data 表中 sales_id1 的记录。, 请帮我删除 sales_id5 的记录。, 请根据最近 10 条销售记录绘制销售额柱状图并返回结果你可以生成并执行 Python 代码。, 请添加一条销售记录product_id3, employee_id2, customer_id7, sale_date2025-10-10, quantity5, amount299.9, discount0.05 ] for q in qs: print(\n\n 用户请求, q) all_chunks [] for chunk in graph.stream({messages: q}, stream_modevalues): all_chunks.append(chunk) if messages in chunk and chunk[messages]: print(- 步骤返回, chunk[messages][-1].content) if all_chunks: last all_chunks[-1] if messages in last and last[messages]: print(最终返回, last[messages][-1].content) if __name__ __main__: print(启动 LangGraph Supervisor) demo_queries() print(示例运行结束。)五、需要注意的坑用LangGraph构建Supervisor系统虽然很高效但也有些问题值得关注。比较常见的一个现象是Supervisor会陷入自言自语的循环。它不断地把一个代理的输出再发给自己来回折腾导致Token消耗暴增响应速度也变慢。还有一个问题是决策的准确性。有时候Supervisor可能会做出不太合理的选择导致系统走错方向。这些问题都需要根据具体的业务逻辑和所用LLM的特性来针对性调整。比如优化system prompt、调整温度参数、或者给Supervisor加上额外的决策规则。六、总结Supervisor 把多代理系统从“各自为政”提升为“有序协作”。它的价值体现在三点结构化的任务调度Supervisor 通过状态控制图实现了任务的动态编排。可扩展的代理体系可以随时接入新的专业代理而无需重构流程。自适应的智能决策通过大模型理解上下文并实时规划下一步。未来多智能体系统的发展方向就是让这些“主管”和“员工”之间的协作更加自然最终演化为一个可以自动理解目标、动态规划路径、自主执行任务的“智能组织”。AI时代未来的就业机会在哪里答案就藏在大模型的浪潮里。从ChatGPT、DeepSeek等日常工具到自然语言处理、计算机视觉、多模态等核心领域技术普惠化、应用垂直化与生态开源化正催生Prompt工程师、自然语言处理、计算机视觉工程师、大模型算法工程师、AI应用产品经理等AI岗位。掌握大模型技能就是把握高薪未来。那么普通人如何抓住大模型风口AI技术的普及对个人能力提出了新的要求在AI时代持续学习和适应新技术变得尤为重要。无论是企业还是个人都需要不断更新知识体系提升与AI协作的能力以适应不断变化的工作环境。因此这里给大家整理了一份《2025最新大模型全套学习资源》包括2025最新大模型学习路线、大模型书籍、视频教程、项目实战、最新行业报告、面试题等带你从零基础入门到精通快速掌握大模型技术由于篇幅有限有需要的小伙伴可以扫码获取1. 成长路线图学习规划要学习一门新的技术作为新手一定要先学习成长路线图方向不对努力白费。这里我们为新手和想要进一步提升的专业人士准备了一份详细的学习成长路线图和规划。2. 大模型经典PDF书籍书籍和学习文档资料是学习大模型过程中必不可少的我们精选了一系列深入探讨大模型技术的书籍和学习文档它们由领域内的顶尖专家撰写内容全面、深入、详尽为你学习大模型提供坚实的理论基础。书籍含电子版PDF3. 大模型视频教程对于很多自学或者没有基础的同学来说书籍这些纯文字类的学习教材会觉得比较晦涩难以理解因此我们提供了丰富的大模型视频教程以动态、形象的方式展示技术概念帮助你更快、更轻松地掌握核心知识。4. 大模型项目实战学以致用当你的理论知识积累到一定程度就需要通过项目实战在实际操作中检验和巩固你所学到的知识同时为你找工作和职业发展打下坚实的基础。5. 大模型行业报告行业分析主要包括对不同行业的现状、趋势、问题、机会等进行系统地调研和评估以了解哪些行业更适合引入大模型的技术和应用以及在哪些方面可以发挥大模型的优势。6. 大模型面试题面试不仅是技术的较量更需要充分的准备。在你已经掌握了大模型技术之后就需要开始准备面试我们将提供精心整理的大模型面试题库涵盖当前面试中可能遇到的各种技术问题让你在面试中游刃有余。为什么大家都在学AI大模型随着AI技术的发展企业对人才的需求从“单一技术”转向 “AI行业”双背景。企业对人才的需求从“单一技术”转向 “AI行业”双背景。金融AI、制造AI、医疗AI等跨界岗位薪资涨幅达30%-50%。同时很多人面临优化裁员近期科技巨头英特尔裁员2万人传统岗位不断缩减因此转行AI势在必行这些资料有用吗这份资料由我们和鲁为民博士(北京清华大学学士和美国加州理工学院博士)共同整理现任上海殷泊信息科技CEO其创立的MoPaaS云平台获Forrester全球’强劲表现者’认证服务航天科工、国家电网等1000企业以第一作者在IEEE Transactions发表论文50篇获NASA JPL火星探测系统强化学习专利等35项中美专利。本套AI大模型课程由清华大学-加州理工双料博士、吴文俊人工智能奖得主鲁为民教授领衔研发。资料内容涵盖了从入门到进阶的各类视频教程和实战项目无论你是小白还是有些技术基础的技术人员这份资料都绝对能帮助你提升薪资待遇转行大模型岗位。大模型全套学习资料已整理打包有需要的小伙伴可以微信扫描下方CSDN官方认证二维码免费领取【保证100%免费】