Building Multi-Agent Supervision Systems: Implementing Innovative AI Solutions and Best Practices with langgraph and langsmith
多智能体系统的兴起：人工智能的第三次浪潮
多智能体系统通常被称为“人工智能的第三次浪潮”。甚至萨蒂亚·纳德拉在一次演讲中也表示，“基于Agent的人工智能将扼杀传统的SaaS应用程序。” https://x.com/MarceloPLima/status/1869389842054025382 听起来很有意思，对吧？
在本文中，我们将深入探讨多智能体系统到底是什么，它们与传统的人工智能工作流程有何不同，以及它们为何会受到如此多的关注。此外，我们将使用 LangGraph 从头开始构建一个多智能体系统，展示其在实际应用中的潜力。
让我们开始吧！
什么是多智能体系统，它们与工作流程有何不同？
在 Anthropic 和 Lex Podcast: Lambert and Dylan Patel 播客中，主持人详细分析了工作流程和多智能体系统之间的区别。
多智能体系统由多个自主智能体组成，这些智能体相互交互、协作或协调以完成任务、解决问题或实现特定目标。这些智能体可以独立运行、动态适应并做出分散的决策。
相比之下，工作流程代表了一系列结构化的任务或流程，旨在实现预定义的结果。与多智能体系统不同，工作流程通常遵循线性或预定义的路径，具有有限的适应性和自主性。
什么是 LangGraph？
LangGraph 是一个旨在促进使用大型语言模型 (LLM) 开发有状态的多智能体应用程序的库。它是 LangChain 生态系统的一部分，通过允许创建循环图来增强基于智能体的工作流程的功能，这些循环图对于建模复杂的交互和工作流程至关重要。
您可以参考这篇文章来了解更多关于 LangGraph 的信息
使用 LangGraph 构建多智能体系统
我们将使用 LangGraph 从头开始构建一个由 LLM 驱动的医生预约系统。该系统将展示多个智能体如何高效协作，无缝处理日程安排、取消和患者互动。
多智能体系统遵循 Supervisor-Worker 模式，将其工作流程组织成三个关键层：
- Planner Layer — 充当编排器，在智能体之间委派任务并确定何时应该结束执行流程。
- Agents Layer — 包含系统中的所有智能体，每个智能体负责特定任务。
- Tooling Layer — 提供一组共享的工具和技能，不同的智能体都可以访问。
现在，让我们逐步深入构建这些层。作为一个实践练习，尝试自己实现一个 FAQ Agent！🚀
导入必要的库和模型
import os, getpass
env_path = 'Path to Env File'
from dotenv import load_dotenv
import json
from langchain_openai import AzureOpenAIEmbeddings, AzureChatOpenAI
load_dotenv(env_path)
from datetime import datetime
from langchain_core.pydantic_v1 import constr, BaseModel, Field, validator
from langgraph.graph import MessagesState
from langchain_core.messages import HumanMessage, SystemMessage,AIMessage
from langchain_core.prompts.chat import ChatPromptTemplate,MessagesPlaceholder
from langchain.pydantic_v1 import BaseModel, Field
from typing_extensions import TypedDict, Annotated
from langgraph.graph import MessagesState, END
from langgraph.types import Command
from langgraph.checkpoint.memory import MemorySaver
from langchain.tools import StructuredTool
from langgraph.graph import START, StateGraph
from langgraph.prebuilt import tools_condition
from langgraph.prebuilt import ToolNode
from langchain_core.runnables import Runnable, RunnableConfig
from langchain_community.tools import TavilySearchResults
from langgraph.prebuilt import create_react_agent
from typing import Literal
from langchain_core.tools import tool
import functools
import pandas as pd
llm = AzureChatOpenAI(temperature=0,
api_key=os.getenv('AZURE_OPENAI_API_KEY'),
azure_endpoint=os.getenv('AZURE_OPENAI_ENDPOINT'),
openai_api_version=os.getenv('AZURE_OPENAI_VERSION'),
azure_deployment=os.getenv('AZURE_GPT4O_MODEL')
)
embeddings = AzureOpenAIEmbeddings(
api_key=os.getenv('AZURE_OPENAI_API_KEY'),
azure_endpoint=os.getenv('AZURE_OPENAI_ENDPOINT'),
azure_deployment=os.getenv('AZURE_OPENAI_EMBEDDINGS_MODEL'),
openai_api_version=os.getenv('AZURE_OPENAI_VERSION'),
)
步骤 1：定义工具层
我们使用 Pydantic 来定义模式
class DateTimeModel(BaseModel):
"""
日期应有的结构和格式
"""
date: str = Field(..., description="格式正确的日期", pattern=r'^\d{4}-\d{2}-\d{2} \d{2}:\d{2}$')
@validator("date")
def check_format_date(cls, v):
if not re.match(r'^\d{4}-\d{2}-\d{2} \d{2}:\d{2}$', v):
raise ValueError("日期应为 'YYYY-MM-DD HH:MM' 格式")
return v
class DateModel(BaseModel):
"""
日期应有的结构和格式
"""
date: str = Field(..., description="格式正确的日期", pattern=r'^\d{2}-\d{2}-\d{4}$')
@validator("date")
def check_format_date(cls, v):
if not re.match(r'^\d{2}-\d{2}-\d{4}$', v):
raise ValueError("日期必须为 'YYYY-MM-DD' 格式")
return v
class IdentificationNumberModel(BaseModel):
"""
ID 应有的结构和格式
"""
id: int = Field(..., description="不带点的识别号", pattern=r'^\d{7,8}$')
@validator("id")
def check_format_id(cls, v):
if not re.match(r'^\d{7,8}$',str(v)):
raise ValueError("ID 号码应为 7 位或 8 位数字")
return v
现在，首先开始为代理定义工具或技能。
@tool
def check_availability_by_doctor(desired_date:DateModel, doctor_name:Literal['kevin anderson','robert martinez','susan davis','daniel miller','sarah wilson','michael green','lisa brown','jane smith','emily johnson','john doe']):
"""
检查数据库中是否有特定医生的空闲时间。
用户应该在查询中提及参数
"""
df = pd.read_csv(f"availability.csv")
df['date_slot_time'] = df['date_slot'].apply(lambda input: input.split(' ')[-1])
rows = list(df[(df['date_slot'].apply(lambda input: input.split(' ')[0]) == desired_date.date)&(df['doctor_name'] == doctor_name)&(df['is_available'] == True)]['date_slot_time'])
if len(rows) == 0:
output = "全天没有空闲时间"
else:
output = f'此时间段 {desired_date.date} 有空闲\n'
output += "空闲时段: " + ', '.join(rows)
return output
@tool
def check_availability_by_specialization(desired_date:DateModel, specialization:Literal["general_dentist", "cosmetic_dentist", "prosthodontist", "pediatric_dentist","emergency_dentist","oral_surgeon","orthodontist"]):
"""
检查数据库中是否有特定专科的空闲时间。
用户应该在查询中提及参数
"""
df = pd.read_csv(f"availability.csv")
df['date_slot_time'] = df['date_slot'].apply(lambda input: input.split(' ')[-1])
rows = df[(df['date_slot'].apply(lambda input: input.split(' ')[0]) == desired_date.date) & (df['specialization'] == specialization) & (df['is_available'] == True)].groupby(['specialization', 'doctor_name'])['date_slot_time'].apply(list).reset_index(name='available_slots')
if len(rows) == 0:
output = "全天没有空闲时间"
else:
def convert_to_am_pm(time_str):
time_str = str(time_str)
hours, minutes = map(int, time_str.split("."))
period = "AM" if hours < 12 else "PM"
hours = hours % 12 or 12
return f"{hours}:{minutes:02d} {period}"
output = f'此时间段 {desired_date.date} 有空闲\n'
for row in rows.values:
output += row[1] + ". 空闲时段: \n" + ', \n'.join([convert_to_am_pm(value)for value in row[2]])+'\n'
return output
@tool
def reschedule_appointment(old_date:DateTimeModel, new_date:DateTimeModel, id_number:IdentificationNumberModel, doctor_name:Literal['kevin anderson','robert martinez','susan davis','daniel miller','sarah wilson','michael green','lisa brown','jane smith','emily johnson','john doe']):
"""
重新安排预约。
用户必须在查询中提及参数。
"""
df = pd.read_csv(f'availability.csv')
available_for_desired_date = df[(df['date_slot'] == new_date.date)&(df['is_available'] == True)&(df['doctor_name'] == doctor_name)]
if len(available_for_desired_date) == 0:
return "在所需时段没有空闲时段"
else:
cancel_appointment.invoke({'date':old_date, 'id_number':id_number, 'doctor_name':doctor_name})
set_appointment.invoke({'desired_date':new_date, 'id_number': id_number, 'doctor_name': doctor_name})
return "已成功重新安排到所需时间"
@tool
def cancel_appointment(date:DateTimeModel, id_number:IdentificationNumberModel, doctor_name:Literal['kevin anderson','robert martinez','susan davis','daniel miller','sarah wilson','michael green','lisa brown','jane smith','emily johnson','john doe']):
"""
取消预约。
用户必须在查询中提及参数。
"""
df = pd.read_csv(f'availability.csv')
case_to_remove = df[(df['date_slot'] == date.date)&(df['patient_to_attend'] == id_number.id)&(df['doctor_name'] == doctor_name)]
if len(case_to_remove) == 0:
return "您没有任何符合这些要求的预约"
else:
df.loc[(df['date_slot'] == date.date) & (df['patient_to_attend'] == id_number.id) & (df['doctor_name'] == doctor_name), ['is_available', 'patient_to_attend']] = [True, None]
df.to_csv(f'availability.csv', index = False)
return "已成功取消"
@tool
def set_appointment(desired_date:DateTimeModel, id_number:IdentificationNumberModel, doctor_name:Literal['kevin anderson','robert martinez','susan davis','daniel miller','sarah wilson','michael green','lisa brown','jane smith','emily johnson','john doe']):
"""
设置与医生的预约或时段。
用户必须在查询中提及参数。
"""
df = pd.read_csv(f'availability.csv')
from datetime import datetime
def convert_datetime_format(dt_str):
dt = datetime.strptime(dt_str, "%Y-%m-%d %H:%M")
return dt.strftime("%d-%m-%Y %#H.%M")
case = df[(df['date_slot'] == convert_datetime_format(desired_date.date))&(df['doctor_name'] == doctor_name)&(df['is_available'] == True)]
if len(case) == 0:
return "没有针对该特定情况的可用预约"
else:
df.loc[(df['date_slot'] == convert_datetime_format(desired_date.date))&(df['doctor_name'] == doctor_name) & (df['is_available'] == True), ['is_available','patient_to_attend']] = [False,
## 步骤 2：定义 Agent 层
在这一层，我们将定义每个专业的 agent，并链接它们对应的工具和指令。
```md
def create_agent(llm:AzureChatOpenAI,tools:list,system_prompt:str):
system_prompt = ChatPromptTemplate.from_messages(
[
(
"system",
system_prompt
),
("placeholder", "{messages}"),
]
)
agent = create_react_agent(model=llm,tools=tools,prompt=system_prompt)
return agent
information_agent = create_agent(
llm=llm,
tools=[check_availability_by_doctor,check_availability_by_specialization],
system_prompt = "您是专门的 agent，用于提供与医生可用性或任何与医院相关的常见问题解答相关的信息。您可以访问该工具。\n 如果您需要任何进一步的信息来执行该工具，请务必礼貌地询问用户。\n 供您参考，请始终将当前年份视为 2024 年。"
)
booking_agent = create_agent(
llm=llm,
tools=[set_appointment,cancel_appointment,reschedule_appointment],
system_prompt = "您是专门的 agent，用于根据查询设置、取消或重新安排预约。您可以访问该工具。\n 如果您需要任何进一步的信息来执行该工具，请务必礼貌地询问用户。\n 供您参考，请始终将当前年份视为 2024 年。"
)
def information_node(state: AgentState):
result = information_agent.invoke(state)
return Command(
update={
"messages": state["messages"] + [
AIMessage(content=result["messages"][-1].content, name="information_node")
]
},
goto="supervisor",
)
def booking_node(state: AgentState):
result = booking_agent.invoke(state)
return Command(
update={
"messages": state["messages"] + [
AIMessage(content=result["messages"][-1].content, name="booking_node")
]
},
goto="supervisor",
)
步骤 3：定义 supervisor 层：
这一层管理 agent 之间的沟通和任务委托。在某些情况下，让 supervisor 为其操作提供推理是有益的，这使得系统更加透明、更容易理解，并且更易于调试。
members_dict = {'information_node':'专门的 agent，用于提供与医生可用性或任何与医院相关的常见问题解答相关的信息。','booking_node':'专门的 agent，仅用于预约、取消或重新安排预约'}
options = list(members_dict.keys()) + ["FINISH"]
worker_info = '\n\n'.join([f'WORKER: {member} \nDESCRIPTION: {description}' for member, description in members_dict.items()]) + '\n\nWORKER: FINISH \nDESCRIPTION: 如果用户查询已回答并路由到完成'
system_prompt = (
"您是一个 supervisor，负责管理以下 workers 之间的对话。 "
"### 专门的助手：\n"
f"{worker_info}\n\n"
"您的主要职责是帮助用户与医生预约，并提供关于常见问题解答和医生可用性的更新。 "
"如果客户要求了解医生的可用性或预约、重新安排或取消预约，"
"请将任务委托给适当的专门 workers。根据以下用户请求，"
"响应要执行的下一个 worker。每个 worker 都将执行一个"
"任务并响应其结果和状态。完成后，"
"响应 FINISH。"
"利用上次对话来评估对话是否应该结束，您是否回答了查询，然后路由到 FINISH "
)
class Router(TypedDict):
"""Worker 用于路由到下一个。如果不需要 workers，则路由到 FINISH。并为路由提供推理"""
next: Annotated[Literal[*options], ..., "worker 用于路由到下一个，路由到 FINISH"]
reasoning: Annotated[str, ..., "支持将路由到 worker 的适当推理"]
def supervisor_node(state: AgentState) -> Command[Literal[*list(members_dict.keys()), "__end__"]]:
messages = [
{"role": "system", "content": system_prompt},
] + [state["messages"][-1]]
query = ''
if len(state['messages'])==1:
query = state['messages'][0].content
response = llm.with_structured_output(Router).invoke(messages)
goto = response["next"]
if goto == "FINISH":
goto = END
if query:
return Command(goto=goto, update={"next": goto,'query':query,'cur_reasoning':response["reasoning"],
"messages":[HumanMessage(content=f"用户的身份识别号是 {state['id_number']}")]
})
return Command(goto=goto, update={"next": goto,'cur_reasoning':response["reasoning"]})
步骤 4：连接所有节点并构建图
builder = StateGraph(AgentState)
builder.add_edge(START, "supervisor")
builder.add_node("supervisor", supervisor_node)
builder.add_node("information_node", information_node)
builder.add_node("booking_node", booking_node)
graph = builder.compile()
步骤 5：对多 agent 系统执行查询
inputs = [
HumanMessage(content='您好，请问您是否可以检查并预约 2024 年 8 月 8 日上午 9 点是否有美容牙医？')
]
config = {"configurable": {"thread_id": "1", "recursion_limit": 10}}
state = {'messages': inputs,'id_number':10232303}
result = graph.invoke(input=state,config=config)
现在要回答这个查询，supervisor 首先要求 ‘availability’ agent 提供在给定日期可用的 ‘美容牙医’ 的姓名，然后调用 ‘booking’ agent，该 agent 调用工具来设置预约，一旦预约完成，supervisor 就结束对话。
此处提供了所有 python 代码和 jupyter notebook
结论：
在本博客中，我们深入研究了多 agent 系统和 LangGraph，探讨了它们的概念和应用。我们逐步介绍了从头开始构建端到端多 agent 系统的过程。
