网站建设英文,深圳专业建设网站,专门做孩子早教的网站,wordpress不能识别语言随着大型语言模型的快速发展#xff0c;构建基于LLM驱动的自治代理#xff08;autonomous agents#xff09;已经成为一个备受关注的话题。仅在过去一年中#xff0c;就出现了许多基于这一理念的新技术和框架。
本文将探索微软开源的Agent框架#xff1a;Autogen… 随着大型语言模型的快速发展构建基于LLM驱动的自治代理autonomous agents已经成为一个备受关注的话题。仅在过去一年中就出现了许多基于这一理念的新技术和框架。
本文将探索微软开源的Agent框架Autogen。它弥补了此类新技术未能解决的空白允许多个Agent协作以实现共同目标。它在LLM之上添加了最小但重要的功能以支持多个代理的初始化和协作允许在多个Agent之间单聊以及群聊。 但作为一个仍处于早期阶段的框架通过API把Autogen集成到实际的生产环境中仍然是一个挑战如Web APP。由于缺乏成熟的文档或资源在Agent通信流中需要一些变通方法。 因此在本文中我们将讨论Autogen Agent连接到API的详细过程。
一、Autogen介绍 在之前的LLM之Agent七| AutoGen介绍文章中介绍过Autogen它是一个基于LLM的Agent通信框架支持创建具有不同人物角色的代理。然后这些代理可以进行单聊也可以进行群聊每个代理轮流发言。
Autogen提供了一些具有不同功能的内置代理类型例如
User Proxy Agent可以检索用户输入并执行代码Assistant Agent默认的系统消息代理该消息允许Agent充当完成任务的助理Conversable Agent在user proxy agents和assistant之间构建会话能力。 虽然Autogen主要支持OpenAI LLM如GPT-3.5和GPT-4来创建代理但用户也可以与本地或其他托管LLM一起使用。 二、Autogen群聊 Autogen中的群聊功能允许多个代理在群设置中进行协作主要特点如下
每个代理都可以看到组中其他代理发送的所有消息一旦启动群聊将继续直到满足其中一个终止条件。例如代理在回复中使用终止消息用户选择退出聊天达到群的最大聊天次数等每个群聊都有一个管理代理负责监督消息广播、发言人选择和聊天终止。Autogen在每轮聊天中选择下一位发言人目前支持四种方法 -manual要求用户手动选择下一个发言人 -random随机选择下一个发言人 -round robin使用循环方法选择下一个发言人 -auto让LLM选择下一个有聊天历史记录作为上下文的发言人 这些特性使Autogen群聊成为Agent协作的理想选择。然而如果想控制Agent在此环境中更多的协作方式时也会带来很多挑战。
三、使用Autogen开发应用程序 目前Autogen旨在作为一种工具使用用户可以完全了解不同代理之间的所有内部通信。这使得将Autogen集成到用户不应该知道这些信息的应用程序中成为一项棘手的工作。 例如如果您构建了一个系统其中多个代理共同担任销售助理那么在决定对用户查询的最终响应之前您可能不想公开他们是如何在内部规划和选择销售策略的。您也可能不想让用户暴露在这种内部沟通的复杂性中。 除此之外在尝试将Autogen代理系统与API集成时我们还面临以下问题
Autogen主要是一个CLI工具。例如它将代理消息打印到CLI并提示用户通过CLI提供反馈Autogen无法在没有明确用户输入的情况下提供一致的方式来结束特定的聊天序列。 但好消息是我们可以使用Autogen已经支持的某些定制来解决这些问题。我们能够将Autogen集成到API中。 下面通过旅游代理系统来演示一下如何将AutoGen与API进行集成
四、基于Autogen的旅游代理系统 该系统将由两个Autogen Assistant Agent和一个User Proxy Agent构建在群聊中这些代理人中的每一位都有以下职责
Tour Agent主代理决定如何响应用户查询以及在生成对用户的最终响应之前应收集的信息Location Researcher旅游代理的助理在通过SERP API查询谷歌地图的函数调用的帮助下进行位置研究。它使代理商能够研究用户心目中与目的地相关的景点、餐厅、住宿等User Proxy代理群聊天中用户的代理。 由于本教程依赖于OpenAI和SERP API因此您需要每个服务的API密钥来尝试本示例。 4.1 Autogen config
首先定义AutoGen的配置
config_list [{model: gpt-3.5-turbo-1106,api_key: os.getenv(OPENAI_API_KEY),}]
4.2 Assistant Agents
然后创建两个助理代理Tour Agent和Location Researcher。 Tour Agent是一个简单的Assistant Agent具有一个自定义的系统提示用于描述其角色和职责它指定代理应如何将TERMINATE添加到针对用户的最终回答的末尾。
tour_agent AssistantAgent(tour_agent, human_input_modeNEVER, llm_config{config_list: config_list,cache_seed: None }, system_messageYou are a Tour Agent who helps users plan a trip based on user requirements. You can get help from the Location Researcher to research and find details about a certain location, attractions, restaurants, accommodation, etc. You use those details a answer user questions, create trip itineraries, make recommendations with practical logistics according to the users requirements. Report the final answer when you have finalized it. Add TERMINATE to the end of this report.) 另一方面在创建Location Researcher时定义一个函数它可以调用并执行来搜索谷歌地图。将在下一节中介绍该函数的实际schema和具体实现。下面代码片段显示了如何通过自定义提示将它们附加到Assistant Agent。
location_researcher AssistantAgent( location_researcher, human_input_modeNEVER, system_messageYou are the location researcher who is helping the Tour Agent plan a trip according to user requirements. You can use the search_google_maps function to retrieve details about a certain location, attractions, restaurants, accommodation, etc. for your research. You process results from these functions and present your findings to the Tour Agent to help them with itinerary and trip planning., llm_config{ config_list: config_list, cache_seed: None, functions: [ SEARCH_GOOGLE_MAPS_SCHEMA, ] }, function_map{ search_google_maps: search_google_maps })
4.3 User Proxy 然后创建User Proxy。尽管User Proxy在该代理系统中没有发挥积极作用但它对于接受用户消息和在向用户发送响应之前检测何时结束对用户查询的回复序列至关重要。
def terminate_agent_at_reply( recipient: Agent, messages: Optional[List[Dict]] None, sender: Optional[Agent] None, config: Optional[Any] None,) - Tuple[bool, Union[str, None]]:return True, Noneuser_proxy UserProxyAgent(user_proxy, is_termination_msglambda x: TERMINATE in x.get(content, ), human_input_modeNEVER, code_execution_configFalse)user_proxy.register_reply([Agent, None], terminate_agent_at_reply) 我已经向User Proxy注册了一个新的回复函数它只返回True和None输出。要了解如何结束聊天序列的必须了解Autogen是如何使用回复功能的。 当代理生成回复时Autogen依赖于注册到代理的回复函数列表。它接受这个列表中的第一个函数如果它能生成最终的回复则返回Truereply。如果返回False表示函数无法生成回复将转移到列表中的下一个函数。 Autogen支持代理不同的回复方式例如请求人工反馈、执行代码、执行函数或生成LLM回复。 当我将terminate_agent_at_reply注册为回复函数时它会被添加到此列表的开头并成为第一个被调用的回复函数。由于默认情况下返回True、None这将阻止用户代理发送自动回复或使用其他回复功能生成LLM回复。使用“None”作为回复会阻止群聊继续进行更多的聊天回合。
4.4 Group chat 最后我将创建允许所有这些代理进行协作的群聊和管理代理。
group_chat GroupChat( agents[self.user_proxy, self.location_researcher, self.tour_agent], messages[], allow_repeat_speakerFalse, max_round20)group_chat_manager GroupChatManager(self.group_chat, is_termination_msglambda x: TERMINATE in x.get(content, ), llm_config{config_list: config_list,cache_seed: None }) 在这里我允许群聊使用默认的发言人选择方法auto因为这个用例没有其他合适的选项。我还设置了聊天管理器的is_terminate_msg参数以检查消息内容中是否存在terminate。 那么当我已经为用户代理使用了以前的terminate_at_agent_reply函数时为什么我要在这里设置另一个终止条件呢 如果LLM在Tour agent的最终回答后选择用户代理以外的代理作为下一个发言人它应该起到故障保护的作用。
4.5 把上述功能合并到一个class里 现在我可以把所有这些代理逻辑放在一个类中还介绍了一种方法来接受来自API的用户消息并在回复序列之后发送最终回复。
import osfrom autogen import AssistantAgent, UserProxyAgent, GroupChat, GroupChatManager, Agentfrom typing import Optional, List, Dict, Any, Union, Callable, Literal, Tuplefrom dotenv import load_dotenvfrom functions import search_google_maps, SEARCH_GOOGLE_MAPS_SCHEMAload_dotenv()config_list [{model: gpt-3.5-turbo-1106,api_key: os.getenv(OPENAI_API_KEY),}]class AgentGroup:def __init__(self): self.user_proxy UserProxyAgent(user_proxy, is_termination_msglambda x: TERMINATE in x.get(content, ), human_input_modeNEVER, code_execution_configFalse ) self.user_proxy.register_reply([Agent, None], AgentGroup.terminate_agent_at_reply) self.location_researcher AssistantAgent(location_researcher, human_input_modeNEVER, system_messageYou are the location researcher who is helping the Tour Agent plan a trip according to user requirements. You can use the search_google_maps function to retrieve details about a certain location, attractions, restaurants, accommodation, etc. for your research. You process results from these functions and present your findings to the Tour Agent to help them with itinerary and trip planning., llm_config{config_list: config_list,cache_seed: None,functions: [ SEARCH_GOOGLE_MAPS_SCHEMA, ] }, function_map{search_google_maps: search_google_maps } ) self.tour_agent AssistantAgent(tour_agent, human_input_modeNEVER, llm_config{config_list: config_list,cache_seed: None }, system_messageYou are a Tour Agent who helps users plan a trip based on user requirements. You can get help from the Location Researcher to research and find details about a certain location, attractions, restaurants, accommodation, etc. You use those details a answer user questions, create trip itineraries, make recommendations with practical logistics according to the users requirements. Report the final answer when you have finalized it. Add TERMINATE to the end of this report. ) self.group_chat GroupChat( agents[self.user_proxy, self.location_researcher, self.tour_agent], messages[], allow_repeat_speakerFalse, max_round20 ) self.group_chat_manager GroupChatManager( self.group_chat, is_termination_msglambda x: TERMINATE in x.get(content, ), llm_config{config_list: config_list,cache_seed: None } )def process_user_message(self, message: str) - str: self.user_proxy.initiate_chat(self.group_chat_manager, messagemessage, clear_historyFalse)return self._find_last_non_empty_message()def _find_last_non_empty_message(self) - str: conversation self.tour_agent.chat_messages[self.group_chat_manager]for i in range(len(conversation) - 1, -1, -1):if conversation[i].get(role) assistant: reply conversation[i].get(content, ).strip() reply reply.replace(TERMINATE, )if reply:return replyreturn No reply received staticmethoddef terminate_agent_at_reply( recipient: Agent, messages: Optional[List[Dict]] None, sender: Optional[Agent] None, config: Optional[Any] None, ) - Tuple[bool, Union[str, None]]:return True, None 这里每当代理组接收到用户消息时用户代理都会启动与组管理器的聊天其中clear_historyFalse可以保留以前回复序列的历史记录。 回复序列结束后find_last_non_empty_message会从聊天记录中找到旅行社代理发送的最后一条非空消息并将其作为答案返回。该函数在寻找应该返回答案时会考虑到与代理回复和回复序列的一些不一致性。
4.6 API endpoint 现在我将创建API端点来接收FastAPI用户查询。
from fastapi import FastAPIfrom pydantic import BaseModelfrom typing import Dictfrom agent_group import AgentGroupclass ChatRequest(BaseModel): session_id: str message: strapp FastAPI()sessions: Dict[str, AgentGroup] {}app.post(/chat)def chat(request: ChatRequest): session_id request.session_id message request.messageif session_id not in sessions.keys(): sessions[session_id] AgentGroup() agent_group sessions[session_id] reply agent_group.process_user_message(message)return {reply: reply, status: success}
4.7 函数调用 最后回过头来处理在前一步中遗漏的内容Location Researcher使用的函数调用的模式和实现。
import osfrom serpapi import GoogleSearchfrom dotenv import load_dotenvfrom typing import Dictload_dotenv()SEARCH_GOOGLE_MAPS_SCHEMA {name: search_google_maps,description: Search google maps using Google Maps API,parameters: {type: object,properties: {query: {type: string,description: A concise search query for searching places on Google Maps } },required: [query] }}def search_google_maps(query): params {engine: google_maps,q: query,type: search,api_key: os.getenv(SERP_API_KEY) } results _search(params) results results[local_results] top_results results[:10] if len(results) 10 else resultsdata []for place in top_results:data.append(_populate_place_data(place[place_id]))return datadef _populate_place_data(place_id: str): params {engine: google_maps,type: place,place_id: place_id,api_key: os.getenv(SERP_API_KEY) }data _search(params)return _prepare_place_data(data[place_results])def _prepare_place_data(place: Dict):return {name: place.get(title),rating: place.get(rating),price: place.get(price),type: place.get(type),address: place.get(address),phone: place.get(phone),website: place.get(website),description: place.get(description),operating_hours: place.get(operating_hours),amenities: place.get(amenities),service_options: place.get(service_options) }def _search(params: Dict[str, str]): search GoogleSearch(params) results search.get_dict()return results search_google_maps函数获取代理提交的搜索查询并将其发送到SERP的google maps API。然后它使用位置ID检索结果中前10个位置的更多详细信息。最后它使用这些详细信息创建一个简化的对象并将其发送回。
4.8 测试 终于到了运行这个应用程序的时候了看看它的效果如何。 我正在尝试一个例子用户发送一个请求来计划去巴厘岛旅行的行程。 Create a week long itinerary to Ubud, Bali for myself for May 2024. I’m going solo and I love exploring nature and going on hikes and activities like that. I have a mid-level budget and I want the itinerary to be relaxing not too packed. 旅行社代理考虑了这一请求并多次联系位置研究员以获取岛上景点、餐厅和住宿选择的详细信息。 旅行社代理根据这些信息提供最终答案。然后用户对这个答案给出更多的反馈使用更多的请求继续进行聊天。
五、改进和定制 在上述的Tour Agent系统中成功地配置了Autogen以更好地适应基于API的应用程序在该应用程序中内部代理通信对用户是隐蔽的。然而它仍然有一些缺点和不可预测的行为。
例如
我假设Tour Agent已经收集了所有所需的详细信息并在用户代理被选为下一位发言人时完成了对用户查询的最终回答。考虑到使用LLM来选择下一个说话者的不可预测性情况可能并不总是如此。之所以选择订单代理是因为发言人并不总是一致的。有时在用户查询后位置研究员会被选为第一个发言人而不是旅行社代理等等。 尽管这些缺点对于这样的用例来说并不重要但对于另一个场景来说可能会有所不同。随着群聊中代理数量的增加这些问题也变得更加明显。 但是如果你想为代理群聊带来更多的可预测性和一致性仍然有一些方法可以定制Autogen的内置行为。
5.1 修改群聊发言人选择提示 Autogen GroupChat类包含一个select_speaker_msg方法可以覆盖该方法以指定如何管理发言人选择。
这是原始提示供参考。
def select_speaker_msg(self, agents: List[Agent]) - str:Return the system message for selecting the next speaker. This is always the *first* message in the context.return fYou are in a role play game. The following roles are available:{self._participant_roles(agents)}.Read the following conversation.Then select the next role from {[agent.name for agent in agents]} to play. Only return the role. 您可以更新此信息具体说明在何种情况下选择哪种代理以使演讲者选择更加一致。
5.2 使用图模型强制执行更严格的说话者转移路径 此笔记本[2]显示了如何将所有允许的说话者转换路径定义为有向图并覆盖GroupChat的select_speaker方法来强制执行该行为。例如对于这个用例可以创建这样一个简单的图。 这些只是几个例子。最终适合您的解决方案将取决于您的特定需求。你还必须在提示和路由方面发挥创意以确保所有代理的行为都符合你的意愿。
参考文献
[1] https://levelup.gitconnected.com/harnessing-the-power-of-autogen-multi-agent-systems-via-api-integration-edb0b9651608
[2] https://github.com/microsoft/autogen/blob/main/notebook/agentchat_graph_modelling_language_using_select_speaker.ipynb
