如何用LangGraph、Qwen和Streamlit構(gòu)建一個(gè)多代理健康助手系統(tǒng)
AI生成的照片
代理是自主代理，能够感知环境并行动以实现特定的目标。在我的先前文章中，我探讨了AI代理的基础知识及其功能。在本文中，我们更进一步，深入探讨一下多代理系统（Multi-Agent Systems，MAS）。
当我们无法用单个智能体处理复杂任务时，我们就会构建多智能体系统，可以将它们定义为一个协同解决问题的智能体网络。存在多种智能体架构，对于我们实现的系统，我们将专注于监督架构。为了驱动我们的智能体，我们将使用Qwen，一个本地的大规模语言模型（LLM）。
在这份教程中，我们将构建一个AI健康助手平台，该平台包含三个代理——健身代理、营养代理和心理健康代理。这三个代理将会协同工作，帮助用户提升体能、营养和身心健康。系统将由一个监督代理协调，该监督代理将任务分配给每个代理并监督他们的进展。
监督架构系统的核心是监督代理，它扮演协调者的角色。它接收用户的指令或输入，识别出适合处理任务各部分的适当代理，并将任务分配给他们。例如，如果用户说：“我想变得健康并且吃得更健康”，监督代理会将此请求交给健身代理和营养师代理。然后，它会收集这些代理的回复，并给出连贯的反馈给用户。在下面的图表中，我们可以看到每个代理如何向监督代理汇报。
现在我们已经讨论了多代理监督系统的概念和架构，让我们来看看具体怎么实现吧。请按照下面的步骤来。
第一步：安装
在我们能够在本地使用Qwen之前，我们需要先安装Ollama，它允许我们在本地机器上直接运行大模型。
1）下载然后安装Ollama，
要下载Ollama，请访问他们的官网并下载适合您操作系统的版本。
2) 检查安装
ollama -v
ii) 获取Qwen模型
ollama pull qwen2.5:14b
我们的系统将会使用外部API来获取真实世界的数据。这些健康助手会从各自的API获取运动和营养信息。
用到的 API:- 健身助手：API-Ninjas 运动 API
- 营养师助手：Spoonacular 食品营养 API
i) 获取你的API密钥
在两个平台上免费注册，然后免费获取你的API密钥。
ii)保存密钥:
为了确保凭证的安全并方便访问，我们将凭证存放在 .env 文件中。
.env文件将会是这样的：
运动API密钥 =xxxxxxxx
饮食API密钥 =xxxxxxxxxxxx当我们构建我们的AI健康助手时，我们需要首先设置的是状态信息。状态信息在帮助代理们记录对话历史方面起着至关重要的作用，特别是在它们在整个流程中互相交流并传递任务时。这样可以帮助代理们记录对话历史。
为了管理这个，我们将使用LangGraph自带的MessagesState类。这个类提供了一种方便的方式来存储和管理消息。我们的自定义状态类将继承MessagesState类以利用其内置功能。
从langchain_core.messages导入HumanMessage, AIMessage
从langgraph.prebuilt导入create_react_agent
从langgraph.graph导入StateGraph, MessagesState, START, END
从langgraph.checkpoint.memory导入MemorySaver
从langchain.prompts导入PromptTemplate
从IPython.display导入display, Image
从typing导入Annotated, Literal
从langchain_ollama导入ChatOllama
从typing_extensions导入TypedDict
从langchain.tools导入tool
从langgraph.types导入Command
导入requests
导入random
导入uuid
导入os
fitness_api_key = os.getenv("EXERCISE_API_KEY")
diet_api_key = os.getenv("DIET_API_KEY")
类State(MessagesState):
next: str之前，我们从API-Ninjas（用于运动数据）和Spoonacular（用于食物和营养数据）获取了API密钥。现在是时候把这些密钥派上用场了，为我们的代理制作自定义工具。这些工具就是代理人完成任务时会用到的。
i)健身器材
我们将使用这个endpoint来获取各种锻炼类型，并为用户制定个性化的锻炼计划。代码如下所示。
class FitnessData:
def __init__(self):
self.base_url = "https://api.api-ninjas.com/v1/exercises"
self.api_key = fitness_api_key
def get_muscle_groups_and_types(self):
muscle_targets = {
'full_body': ["abdominals", "biceps", "calves", "chest", "forearms", "glutes",
"hamstrings", "lower_back", "middle_back", "quadriceps",
"traps", "triceps", "adductors"
],
'upper_body': ["biceps", "chest", "forearms", "lats", "lower_back", "middle_back", "neck", "traps", "triceps" ],
'lower_body': ["adductors", "calves", "glutes", "hamstrings", "quadriceps"]
}
exercise_types = {'types':["powerlifting","strength", "stretching", "strongman"]}
return muscle_targets, exercise_types
def fetch_exercises(self, type, muscle, difficulty):
headers = {
'X-Api-Key':self.api_key
}
params= {
'type': type,
'muscle': muscle,
'difficulty': difficulty
}
try:
response = requests.get(self.base_url, headers=headers,params=params)
result = response.json()
if not result:
print(f"没有找到针对 {muscle} 的运动")
return result
except requests.RequestException as e:
print(f"请求失败：{e}")
return []
def generate_workout_plan(self, query='full_body', difficulty='intermediate'):
output=[]
muscle_targets, exercise_types = self.get_muscle_groups_and_types()
muscle = random.choice(muscle_targets.get(query))
type = random.choice(exercise_types.get('types'))
result = self.fetch_exercises('拉伸', muscle, difficulty)
print(result)
limit_plan = result[:3]
for i, data in enumerate(limit_plan):
if data not in output:
output.append(f"运动 {i+1}：{data['name']}")
output.append(f"锻炼部位：{data['muscle']}")
output.append(f"说明：{data['instructions']}")
return output之后，我们通过创建类的实例来调用 generate_workout_plan 函数，从而建立了一个健身自定义工具。此函数允许用户根据特定类别（比如全身、上半身或下半身）来定制锻炼计划。请注意，该函数带有 @tool 装饰器；这个装饰器就是将函数转变成 LangChain 自定义工具的关键。
@工具注解
def fitness_data_tool(query: Annotated[str, "此输入应为 full_body、upper_body 或 lower_body 中的一种"]):
"""此工具用于为用户提供健身或锻炼计划。
输入的锻炼类型（full_body、upper_body 或 lower_body）将作为您的输入。
"""
fitness_tool = FitnessData()
result = fitness_tool.generate_workout_plan(query)
return result饮食营养师工具
对于营养师助手的数据来源，我们将利用Spoonacular API，并利用其生成饮食计划和获取食谱信息端点。通过这种方式，助手可以根据用户的饮食偏好（如素食、纯素食或标准饮食等）生成个性化的饮食计划。用户可以看到包含每日营养成分明细（如蛋白质、脂肪和碳水化合物）的饮食计划。
class 营养师类:
def __init__(self):
self.base_url = "https://api.spoonacular.com"
self.api_key = diet_api_key
def 获取餐食(self, 时间范围="day", 饮食="无"):
url = f"{self.base_url}/mealplanner/generate"
params = {
"timeFrame": 时间范围,
"diet": 饮食,
"apiKey": self.api_key
}
response = requests.get(url, params=params)
if not response:
print('未找到餐计划，请检查输入参数')
return response.json()
def 获取食谱信息(self, 食谱ID):
url = f"{self.base_url}/recipes/{食谱ID}/information"
params = {"apiKey": self.api_key}
response = requests.get(url, params=params)
if not response:
print("未找到食谱，请检查 API 密钥和输入参数")
return response.json()
def 生成餐计划(self, 查询):
处理餐食 = []
餐计划 = self.获取餐食(查询)
print(餐计划)
餐食 = 餐计划.get('meals')
营养成分 = 餐计划.get('nutrients')
for i, 餐食项 in enumerate(餐食):
食谱信息 = self.获取食谱信息(餐食项.get('id'))
食材 = [食材['original'] for 食材 in 食谱信息.get('extendedIngredients')]
处理餐食.append(f"🍽️ 餐点 {i+1}: {餐食项.get('title')}")
处理餐食.append(f"准备时间 (分钟): {餐食项.get('readyInMinutes')}")
处理餐食.append(f"份量 (人): {餐食项.get('servings')}")
处理餐食.append("食材:\n" + "\n".join(食材))
处理餐食.append(f"指南:\n {食谱信息.get('instructions')}")
处理餐食.append(
"\n每日营养:\n"
f"蛋白质: {营养成分.get('protein', 'N/A')} g\n"
f"脂肪: {营养成分.get('fat', 'N/A')} g\n"
f"碳水化合物: {营养成分.get('carbohydrates', 'N/A')} g"
)
return 处理餐食
接下来，我们来创建自己的工具。
@工具装饰器
def diet_tool(query: Annotated[str, "输入可以是 None、素食或纯素食"]):
"""使用此工具为用户提供饮食计划。
输入的饮食类型会生成相应的饮食计划
"""
营养师工具 = Dietitian()
结果 = 营养师工具.generate_meal_plan(query)
return result在这里，我们将定义这个大型语言模型，名为Qwen2.5:14b。该模型非常适合用于构建智能助手。
llm = ChatOllama(model="qwen2.5:14b") # 创建一个名为llm的ChatOllama对象，使用模型"qwen2.5:14b"
memory = MemorySaver() # 创建一个名为memory的MemorySaver对象在这一步，我们将创建节点以及代理，我们将使用 LangGraph 中预构建的create_react_agent工具。
i) 健身教练对于健身代理程序，我们将以下三个关键组件传递给 create_react_agent 函数：LLM、我们自定义的用于获取运动数据的工具（fitness_data_tool）和健身代理提示（fitness_agent_prompt）。
接下来，在我们的健身任务节点中，该节点代表LangGraph工作流中的健身任务。我们通过传递当前状态中的消息（即用户的输入信息）来调用代理。处理完成后，我们使用命令对象传递结果。这让我们能够用输出更新状态，并指示其返回主管代理。一旦任务完成，我们就指示健身节点返回主管代理。
fitness_agent_prompt = """
你只能回答关于健身的问题。
"""
fitness_agent = create_react_agent(
llm,
tools = [fitness_data_tool],
prompt = fitness_agent_prompt)
def fitness_node(state: State) -> Command[Literal["supervisor"]]:
result = fitness_agent.invoke(state)
return Command(
update={
"messages": [
AIMessage(content=result["messages"][-1].content, name="健身消息")
]
},
goto="supervisor",
)ii) 营养师代理
在创建营养师代理及其节点的过程中，我们遵循了同样的步骤。
dietitian_system_prompt = """
你只能回答关于饮食和餐计划安排的问题。
"""
dietitian_agent = create_react_agent(
llm,
tools = [diet_tool],
prompt = dietitian_system_prompt)
def dietitian_node(state: State) -> Command[Literal["supervisor"]]:
result = dietitian_agent.invoke(state)
return Command(
update={
"messages": [
AIMessage(content=result["messages"][-1].content, name="dietitian")
]
},
goto="supervisor",
)iii) 心理健康辅导员
为了创建我们的心理健康代理，我们定义了一个 mental_health_node，它包含一个自定义提示来引导大语言模型完成任务，并告知其预期结果。任务完成后，该节点使用 Command 对象来更新对话状态以反映当前情况，然后将控制权交回 Supervisor Agent。
def 心理健康节点(state: State) -> Command[Literal["supervisor"]]:
提示模板 = PromptTemplate.from_template(
"""你是一位支持性的心理健康教练。
你的任务是：
- 提供一个独特的心理健康建议或减压练习。
- 使其简单、亲切且实用。避免重复建议。"""
)
链 = 提示模板 | llm
回复 = 链.invoke(state)
return Command(
update={
"messages": [
AIMessage(content=f"这是你的健康贴士：{回复.content}", name="wellness")
]
},
goto="supervisor",
)iv) 主管代理
在创建监督代理的过程中，我们定义了一个系统提示，在其中明确代理的角色并介绍它将管理的团队成员，包括健身代理、营养师代理和心理健康代理这三名成员。我们还定义了一个路由器类，用以规范监督代理的输出，作为监督代理输出的结构化模板。
然后，我们实现了监控节点，设置了消息的流程，并定义了代理之间路由的消息逻辑。这包括如何将消息路由到下一个任务以及何时结束会话。
members = ["fitness", "dietitian", "wellness"]
options = members + ["FINISH"]
system_prompt = (
"你是负责管理以下员工之间的对话的主管：\n"
f"{members}。在收到用户的请求后，\n"
"请回复接下来应由哪个员工行动。每个员工将执行一项任务并回复他们的结果和状态。任务完成后，\n"
"请回复 FINISH。\n"
"指南：\n"
"1. 请检查对话中的最后一条消息，确定任务是否已完成。\n"
"2. 如果您已经得到了最终答案或结果，请回复 'FINISH'。\n"
)
class Router(TypedDict):
"""如果没有需要处理的工作人员，请回复 FINISH。"""
next: Literal[*options]
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})现在，我们构建工作流图，将主管节点作为执行的起点。接着，我们再添加其他的代理节点。
builder = StateGraph(状态)
builder.add_edge(开始, "supervisor")
builder.add_node("supervisor", 监督节点) # 监督节点指的是负责监督的节点
builder.add_node("fitness", 健身节点) # 健身节点指的是与健身相关的节点
builder.add_node("dietitian", 营养师节点) # 营养师节点指的是与营养师相关的节点
builder.add_node("wellness", 心理健康节点) # 心理健康节点指的是与心理健康相关的节点
graph = builder.compile(checkpointer=memory)
在这个阶段，我们的多代理系统已经完全准备好，准备好接收用户的输入。发送输入之前，我们先定义一个帮助函数来提取代理的输出。
def 解析语言图输出(stream): # 解析语言图输出
结果 = [] # 结果列表
for 键, 值 in stream.items(): # 遍历键值对
if 键 == "supervisor": # 如果键是"supervisor"，继续下一个循环
continue
消息 = 值.get("messages", []) # 获取消息列表
for 消息项 in 消息: # 遍历消息列表
if isinstance(消息项, str): # 判断是否为字符串
结果.append((键, 消息项)) # 添加键值对到结果列表
elif isinstance(消息项, AIMessage): # 判断是否为AIMessage
结果.append((键, 消息项.content)) # 添加键值对到结果列表
return 结果 # 返回结果列表我们将用户的输入输入到系统中。
# 获取流的最终事件
final_event = None
config = {"configurable": {"thread_id": "1", "recursion_limit": 10}}
inputs = {
"messages": [
HumanMessage(
content="给我一些关于这个月的健康建议?"
)
],
}
for step in graph.stream(inputs, config=config):
final_event = step # 始终使用最新的事件
输出当前事件：final_event
response_message = parse_langgraph_output(final_event)
for agent, content in response_message:
显示 "**Agent :** `{agent}`\n\n{content}"
显示 "="*50这是在Streamlit应用中显示的结果。
查看这个GitHub仓库的完整代码。
谢谢你的阅读！下回见,。
共同學(xué)習(xí)，寫(xiě)下你的評(píng)論
評(píng)論加載中...
作者其他優(yōu)質(zhì)文章
100積分直接送
付費(fèi)專(zhuān)欄免費(fèi)學(xué)
大額優(yōu)惠券免費(fèi)領(lǐng)