智能体监督者

🌟 多智能体监督员：让AI团队协作更高效！🌟

大家好！今天我们要聊的是一个超级酷炫的话题——多智能体监督员！想象一下，你有一支AI团队，每个成员都有自己擅长的领域，比如有的擅长做研究，有的擅长写代码。那么，如何让这些AI成员高效协作，完成复杂的任务呢？这就是我们今天要探讨的内容！

🛠️ 准备工作

首先，我们需要安装一些必要的工具包，并设置API密钥。别担心，跟着我一步步来，你也能轻松搞定！

python
%%capture --no-stell
%pip install -U langgraph langchain_community langchain_anthropic langchain_experimental

import getpass
import os

def _set_if_undefined(var: str):
    if not os.environ.get(var):
        os.environ[var] = getpass.getpass(f"请提供你的 {var}")

_set_if_undefined("ANTHROPIC_API_KEY")
_set_if_undefined("TAVILY_API_KEY")

🛠️ 创建工具

接下来，我们需要为我们的AI团队创建一些工具。比如，一个用于网络搜索的工具，另一个用于生成图表的工具。

python
from typing import Annotated
from langchain_community.tools.tavily_search import TavilySearchResults
from langchain_core.tools import tool
from langchain_experimental.utilities import PythonREPL

tavily_tool = TavilySearchResults(max_results=5)

# 这个工具可以执行Python代码，但要注意安全哦！
repl = PythonREPL()

@tool
def python_repl_tool(
    code: Annotated[str, "要执行的Python代码，用于生成图表。"],
):
    """使用这个工具来执行Python代码并进行数学计算。如果你想看到某个值的输出，记得用`print(...)`打印出来哦！"""
    try:
        result = repl.run(code)
    except BaseException as e:
        return f"执行失败。错误：{repr(e)}"
    result_str = f"成功执行：\n\`\`\`python\n{code}\n\`\`\`\n标准输出：{result}"
    return result_str

🕵️‍♂️ 创建智能体监督员

现在，我们需要一个“监督员”来协调这些AI成员的工作。这个监督员会使用LLM（大语言模型）来决定下一个该由谁来处理任务，或者任务是否已经完成。

python
from typing import Literal
from typing_extensions import TypedDict
from langchain_anthropic import ChatAnthropic
from langgraph.graph import MessagesState, END
from langgraph.types import Command

members = ["researcher", "coder"]
options = members + ["FINISH"]

system_prompt = (
    "你是一个监督员，负责管理以下成员之间的对话：{members}。根据用户请求，决定下一个该由谁来处理任务。每个成员会执行任务并返回结果和状态。当任务完成时，请返回FINISH。"
)

class Router(TypedDict):
    """决定下一个该由谁来处理任务。如果不需要其他成员处理，返回FINISH。"""
    next: Literal[*options]

llm = ChatAnthropic(model="claude-3-5-sonnet-latest")

class State(MessagesState):
    next: str

def supervisor_node(state: State) -> Command[Literal[*members, "__end__"]]:
    messages = [
        {"role": "system", "content": system_prompt},
    ] + state["messages"]
    response = llm.with_structured_output(Router).invoke(messages)
    goto = response["next"]
    if goto == "FINISH":
        goto = END

    return Command(goto=goto, update={"next": goto})

🏗️ 构建图

接下来，我们开始构建图。定义状态和各个工作节点。

python
from langchain_core.messages import HumanMessage
from langgraph.graph import StateGraph, START, END
from langgraph.prebuilt import create_react_agent

research_agent = create_react_agent(
    llm, tools=[tavily_tool], prompt="你是一个研究员。不要做任何数学计算。"
)

def research_node(state: State) -> Command[Literal["supervisor"]]:
    result = research_agent.invoke(state)
    return Command(
        update={
            "messages": [
                HumanMessage(content=result["messages"][-1].content, name="researcher")
            ]
        },
        goto="supervisor",
    )

# 注意：这个工具可以执行任意代码，使用时要注意安全！
code_agent = create_react_agent(llm, tools=[python_repl_tool])

def code_node(state: State) -> Command[Literal["supervisor"]]:
    result = code_agent.invoke(state)
    return Command(
        update={
            "messages": [
                HumanMessage(content=result["messages"][-1].content, name="coder")
            ]
        },
        goto="supervisor",
    )

builder = StateGraph(State)
builder.add_edge(START, "supervisor")
builder.add_node("supervisor", supervisor_node)
builder.add_node("researcher", research_node)
builder.add_node("coder", code_node)
graph = builder.compile()

🚀 启动团队

现在，我们的AI团队已经准备就绪，可以开始执行任务了！

python
for s in graph.stream(
    {"messages": [("user", "42的平方根是多少？")]}, subgraphs=True
):
    print(s)
    print("----")

🎉 总结

通过这个教程，我们学会了如何创建一个多智能体监督员，让AI团队成员高效协作，完成复杂的任务。希望这个教程对你有所帮助，快去试试吧！如果你有任何问题或想法，欢迎在评论区留言哦！😄

小贴士：在使用Python REPL工具时，一定要注意代码的安全性，避免执行恶意代码哦！🔒

Happy coding! 🚀