如何用AI進(jìn)行情感分析
與此同時(shí)，也有越來(lái)越多的Agent框架開始聚焦于multi-agent場(chǎng)景。為了完成任務(wù)，multi-agent會(huì)為不同的Agent賦予不同的角色定位，通過(guò)Agent之間的協(xié)同合作來(lái)完成復(fù)雜的任務(wù)。而在完成任務(wù)的過(guò)程中，相比于single-agent來(lái)說(shuō)，與用戶的交互會(huì)更少一些。
為了構(gòu)建一個(gè)multi-agent框架，我們需要思考相比于single-agent，框架中多了哪些組件。
multi-agent框架的核心交互流程可以概括如下：
下面的部分我們將簡(jiǎn)單介紹現(xiàn)有的三個(gè)multi-agent框架，并對(duì)它們進(jìn)行簡(jiǎn)單的比較。
相關(guān)資料
代碼repo：https://github.com/geekan/MetaGPT
論文：https://arxiv.org/pdf/2308.00352.pdf核心模塊
MetaGPT論文中給出的架構(gòu)圖如上所示。
class Role:
"""Role/Agent"""
def __init__(self, name="", profile="", goal="", constraints="", desc=""):
self._llm = LLM() #llm
self._setting = RoleSetting(name=name, profile=profile, goal=goal, constraints=constraints, desc=desc)
self._states = []
self._actions = [] # 對(duì)應(yīng)的action/ stage
self._role_id = str(self._setting)
self._rc = RoleContext()在Role中，又有兩個(gè)主要組件：RoleSetting和RoleContext。其中RoleSetting存儲(chǔ)了角色的名字、目標(biāo)等信息。RoleContext則包含了運(yùn)行中的狀態(tài)信息，如記憶，需要執(zhí)行的動(dòng)作等。
class RoleSetting(BaseModel):
"""Role Settings"""
name: str
profile: str
goal: str
constraints: str
desc: str
class RoleContext(BaseModel):
"""Role Runtime Context"""
# 維護(hù)運(yùn)行過(guò)程中的一些狀態(tài)信息
env: 'Environment' = Field(default=None) # environment，所有角色共享
memory: Memory = Field(default_factory=Memory) # 記憶信息
state: int = Field(default=0) # 下一時(shí)刻要執(zhí)行的動(dòng)作
todo: Action = Field(default=None)
watch: set[Type[Action]] = Field(default_factory=set) # 關(guān)注/訂閱的信息
news: list[Type[Message]] = Field(default=[]) # 本輪新增的信息下面簡(jiǎn)單介紹一下MetaGPT的流程。
首先，在environment.run()函數(shù)中，會(huì)依次調(diào)用每個(gè)角色的role.run()
async def run(self, k=1):
"""處理一次所有信息的運(yùn)行
Process all Role runs at once
"""
for _ in range(k):
futures = []
# 執(zhí)行
for role in self.roles.values():
future = role.run()
futures.append(future)
await asyncio.gather(*futures)在role.run()函數(shù)中，與前面所述流程類似分為三步：a) _observe()函數(shù)觀察環(huán)境、更新memory。b) _react()函數(shù)思考并執(zhí)行動(dòng)作。c) _publish_message()函數(shù)將執(zhí)行結(jié)果更新到環(huán)境中。
async def run(self, message=None):
"""Observe, and think and act based on the results of the observation"""
if message:
...
# 觀察環(huán)境、更新memory
elif not await self._observe():
# If there is no new information, suspend and wait
return
# 執(zhí)行動(dòng)作
rsp = await self._react()
# 把結(jié)果傳給environment
self._publish_message(rsp)
return rsp_react()函數(shù)可以進(jìn)一步分為_think()和_act()兩個(gè)階段。_think()負(fù)責(zé)決定SOP的狀態(tài)轉(zhuǎn)移，_act()通過(guò)調(diào)用action.run()完成具體執(zhí)行。兩個(gè)階段可能都需要調(diào)用LLM。
自定義擴(kuò)展
最后在作者給出的examples中，用戶如果想要實(shí)現(xiàn)一個(gè)自定義場(chǎng)景，一般情況下只需要實(shí)現(xiàn)自己的Action和Role類及對(duì)應(yīng)的關(guān)鍵函數(shù)即可。
相關(guān)資料
代碼repo：https://github.com/OpenBMB/AgentVerse
論文：https://arxiv.org/pdf/2308.10848.pdf除此之外，清華大學(xué)的開源框架AgentVerse也提供了一個(gè)基于LLMs來(lái)搭建多智能體交互的框架。
核心模塊
論文中，提供了如上圖所示的流程圖，它的設(shè)計(jì)流程模擬了人類的決策過(guò)程，包含了四個(gè)階段，分別是：專家招聘、協(xié)作決策、行動(dòng)執(zhí)行和評(píng)估。
根據(jù)包括軟件開發(fā)，咨詢和游戲的幾個(gè)不同領(lǐng)域的案例表明，這種多智能體合作的方式優(yōu)于單一智能體，并且優(yōu)勢(shì)明顯。
AgentVerse 在代碼實(shí)現(xiàn)上，包含了兩個(gè)基礎(chǔ)元素agent和environment， agent不用過(guò)多介紹，他的工作行為依賴特定的prompt以及配合大模型llm進(jìn)行結(jié)果生成。
這里的environment就是代表著任務(wù)，在environment中通過(guò)定義agents以及rules等信息，來(lái)確保多智能體的交互流程能夠按照上述進(jìn)行。
值得注意的是，environment 的rules中包含了5個(gè)組建，這些組件分別是： 描述器，順序，選擇器，更新器，以及可見(jiàn)性。他們的作用具體如下，（參考https://mp.weixin.qq.com/s/jkW2JRnbfsK81ClhwsCxqA）
具體Environment定義的代碼如下：
@EnvironmentRegistry.register("basic")
class BasicEnvironment(BaseEnvironment):
"""
A basic environment implementing the logic of conversation.
Args:
agents: List of agents
rule: Rule for the environment
max_turns: Maximum number of turns
cnt_turn: Current turn number
last_messages: Messages from last turn
rule_params: Variables set by the rule
"""
agents: List[BaseAgent]
rule: Rule
max_turns: int = 10
cnt_turn: int = 0
last_messages: List[Message] = []
rule_params: Dict = {}
def __init__(self, rule, **kwargs):
rule_config = rule
order_config = rule_config.get("order", {"type": "sequential"})
visibility_config = rule_config.get("visibility", {"type": "all"})
selector_config = rule_config.get("selector", {"type": "basic"})
updater_config = rule_config.get("updater", {"type": "basic"})
describer_config = rule_config.get("describer", {"type": "basic"})
rule = Rule(
order_config,
visibility_config,
selector_config,
updater_config,
describer_config,
)
super().__init__(rule=rule, **kwargs)對(duì)應(yīng)Agent的定義如下：
class BaseAgent(BaseModel):
name: str # 名字
llm: BaseLLM # llm
output_parser: OutputParser # 輸出解析
prompt_template: str # 模板
role_description: str = Field(default="") # 角色信息
memory: BaseMemory = Field(default_factory=ChatHistoryMemory) # 記憶
max_retry: int = Field(default=3)
receiver: Set[str] = Field(default=set({"all"})) # 信息的接收方
async_mode: bool = Field(default=True)核心流程
用戶初始化，并調(diào)用 agentVerse.run()作為程序入口
class AgentVerse:
def __init__(self, agents: List[BaseAgent], environment: BaseEnvironment):
self.agents = agents
self.environment = environment
def run(self):
"""Run the environment from scratch until it is done."""
self.environment.reset()
# 循環(huán)執(zhí)行
while not self.environment.is_done():
asyncio.run(self.environment.step())相關(guān)environment流程邏輯如下：
async def step(self) -> List[Message]:
"""Run one step of the environment"""
# 選擇下一時(shí)刻的行動(dòng)的agent(s)
agent_ids = self.rule.get_next_agent_idx(self)
# 環(huán)境描述信息（每個(gè)agents不一定一樣）
env_descriptions = self.rule.get_env_description(self)
# agent行動(dòng)，返回結(jié)果
messages = await asyncio.gather(
*[self.agents[i].astep(env_descriptions[i]) for i in agent_ids]
)
# 選擇過(guò)濾message信息
selected_messages = self.rule.select_message(self, messages)
# 更新mmory
self.rule.update_memory(self)
# 更新agent之間的可見(jiàn)性
self.rule.update_visible_agents(self)
self.cnt_turn += 1
return selected_messages相關(guān) agent 流程邏輯如下：
def step(self, env_description: str = "") -> Message:
parsed_response = None
tool_observation = [self.tool_memory.to_string()]
while True:
# 拼接prompt
prompt = self._fill_prompt_template(env_description, tool_observation)
try:
# 調(diào)用LLM
response = self.llm.generate_response(prompt)
# 解析結(jié)果
parsed_response = self.output_parser.parse(response)
if isinstance(parsed_response, AgentAction):
# 調(diào)用工具
observation = self._call_tool(parsed_response)
tool_observation.append(
parsed_response.log.strip()
+ f"\nObservation: {observation.strip()}"
)
break
except BaseException as e:
logging.error(e)
logging.warning("Retrying...")
continue
if parsed_response is None or isinstance(parsed_response, AgentFinish):
break
self._update_tool_memory(tool_observation)
message = Message(
content=""
if parsed_response is None
else parsed_response.return_values["output"],
sender=self.name,
receiver=self.get_receiver(),
)
return message拿軟件開發(fā)的任務(wù)舉例子，用戶需要提供一個(gè)json配置文件，包含environment信息所需的信息，并且定義若干個(gè)不同的agent，如code_writer，code_reviwer，unit_test_generator，以及他們之間溝通時(shí)候的rule，最終初始化environment， agents以及agentverse入口方法，并調(diào)用 agentVerse.run()即可。
相關(guān)資料
代碼repo：https://github.com/aiwaves-cn/agents
論文：https://arxiv.org/pdf/2309.07870.pdf最后，波形智能聯(lián)合浙大和蘇黎世聯(lián)邦理工大學(xué)，提供了另外一種支持mutli-agent的設(shè)計(jì)方案。
核心模塊
該方案圍繞SOP模塊來(lái)確定agent之間交互的推進(jìn)方式，通過(guò)SOP來(lái)管理狀態(tài)的變換，并將相關(guān)狀態(tài)信息變換記錄到環(huán)境中，以便不同的agent進(jìn)行各自任務(wù)的推進(jìn)。 整體流程如下圖所示：
Agents有三個(gè)核心模塊，除了上述SOP，還有Agent以及Environment。
（參考：https://mp.weixin.qq.com/s/toblMJJkpFKtv0dfJFfHKA）
核心流程
入口代碼如下所示，用戶通過(guò)配置文件來(lái)初始化agent，environment以及sop，并通過(guò)如下入口程序開始agent交互流程。
def run(agents,sop,environment):
while True:
# 更新狀態(tài)，決定下一個(gè)行動(dòng)的agent
current_state,current_agent= sop.next(environment,agents)
if sop.finished:
os.environ.clear()
break
# agent執(zhí)行
action = current_agent.step(current_state)
# 更新memory
memory = process(action)
environment.update_memory(memory,current_state)sop的流程方法如下，通過(guò)調(diào)用transit方法進(jìn)行狀態(tài)轉(zhuǎn)移的操作，以及用route進(jìn)行agent的選擇。
def next(self, environment, agents):
"""
Determine the next state and the agent that needs action based on the current situation
"""
# 一些初始化流程：獲取記憶、相關(guān)信息等
...
# 下一時(shí)刻的狀態(tài)
next_state = self.transit(
chat_history=environment.shared_memory["long_term_memory"][
environment.current_chat_history_idx :
],
relevant_history=relevant_history,
environment=environment,
)
# 如果進(jìn)入終止節(jié)點(diǎn)，則直接終止
if next_state.name == self.finish_state_name:
self.finished = True
return None, None
# 更新狀態(tài)
self.current_state = next_state
# 決定下一時(shí)刻要行動(dòng)的agent
next_agent = self.route(
chat_history=environment.shared_memory["long_term_memory"][
environment.current_chat_history_idx :
],
agents = agents,
relevant_history=relevant_history,
)
return self.current_state, next_agentagent的執(zhí)行方法如下，主要是根據(jù)當(dāng)前環(huán)境信息，來(lái)進(jìn)行prompt生成，并調(diào)用llm進(jìn)行生成
def step(self, current_state,input=""):
"""
return actions by current state and environment
Return: action(Action)
"""
current_state.chat_nums +=1
state_begin = current_state.is_begin
agent_begin = self.begins[current_state.name]["is_begin"]
self.begins[current_state.name]["is_begin"] = False
current_state.is_begin = False
environment = self.environment
self.current_state = current_state
# 先根據(jù)當(dāng)前環(huán)境更新信息
if len(environment.shared_memory["long_term_memory"])>0:
current_history = self.observe()
self.long_term_memory.append(current_history)
response,res_dict = self.act()
action_dict = {
"response": response,
"res_dict": res_dict,
"role": self.state_roles[current_state.name],
"name": self.name,
"state_begin" : state_begin,
"agent_begin" : agent_begin,
"is_user" : self.is_user
}
return Action(**action_dict)
def act(self):
"""
return actions by the current state
"""
current_state = self.current_state
chat_history = self.long_term_memory
current_LLM = self.LLMs[current_state.name]
# 拼接prompt
system_prompt, last_prompt, res_dict = self.compile()
# 調(diào)用LLM
response = current_LLM.get_response(
chat_history, system_prompt, last_prompt, stream=True
)
return response,res_dict最后，agent執(zhí)行結(jié)束后，environment會(huì)更新本輪環(huán)境信息用于下一輪
class Memory:
def __init__(self,role,name,content) -> None:
self.send_role = role
self.send_name = name
self.content = content
def update_memory(self, memory, current_state):
"""
更新環(huán)境的memory信息
"""
MAX_CHAT_HISTORY = eval(os.environ["MAX_CHAT_HISTORY"])
self.shared_memory["long_term_memory"].append(memory)
current_embedding = get_embedding(memory.content)
# 對(duì)過(guò)去幾輪信息作summary
if len(self.shared_memory["long_term_memory"]) % MAX_CHAT_HISTORY == 0:
summary = self.summary(current_state)
self.shared_memory["short_term_memory"] = summary
self.agents[memory.send_name].update_memory(memory)目前，Agents 除了獨(dú)特的SOP系統(tǒng)以外，相對(duì)于其他框架不同的點(diǎn)還有一個(gè)重要功能是支持人類使用者扮演multi-agent系統(tǒng)中的一個(gè)或多個(gè)智能體的功能，可以方便地支持各種人 – 智能體交互的應(yīng)用場(chǎng)景，如人和智能體一起玩游戲、辯論等。
最后，我們針對(duì)上述幾個(gè)市面上比較主流的agents框架進(jìn)行簡(jiǎn)單的總結(jié)與比較。部分相關(guān)特性如下所示：
| 框架名稱 | MetaGPT | AgentVerse | Agents |
| Config的詳細(xì)性 | 少 | 中 | 多 |
| 狀態(tài)的切換 | 基于LLM與SOP | 無(wú)/迭代達(dá)到最大輪數(shù) | 基于LLM與SOP |
| Agent執(zhí)行順序 | 順序 | 基于規(guī)則 | 順序/LLM決定/隨機(jī) |
| 長(zhǎng)短期記憶 | 有 | 有 | 有/做了額外處理 |
| 工具調(diào)用 | 少 | 少/單個(gè) | 少/不是由LLM決定的 |
| Agent差異性體現(xiàn)（即對(duì)當(dāng)前Agent，其他Agent是否有區(qū)別） | 有 | 有 | 無(wú) |
綜上，multi-agent框架的核心還是在agent之間的交互與狀態(tài)的轉(zhuǎn)移。而LLM對(duì)工具的調(diào)用往往只是作為其中一個(gè)子模塊，且大部分agent都只調(diào)用單個(gè)tool。因此，在multi-agent的場(chǎng)景中，使用多個(gè)單tool的agent還是一個(gè)多tool的agent，也是一個(gè)值得考慮的問(wèn)題。
本文章轉(zhuǎn)載微信公眾號(hào)@青稞AI
對(duì)比大模型API的內(nèi)容創(chuàng)意新穎性、情感共鳴力、商業(yè)轉(zhuǎn)化潛力
一鍵對(duì)比試用API 限時(shí)免費(fèi)
