如何用AI進行情感分析
與此同時，也有越來越多的Agent框架開始聚焦于multi-agent場景。為了完成任務，multi-agent會為不同的Agent賦予不同的角色定位，通過Agent之間的協同合作來完成復雜的任務。而在完成任務的過程中，相比于single-agent來說，與用戶的交互會更少一些。
為了構建一個multi-agent框架，我們需要思考相比于single-agent，框架中多了哪些組件。
multi-agent框架的核心交互流程可以概括如下：
下面的部分我們將簡單介紹現有的三個multi-agent框架，并對它們進行簡單的比較。
相關資料
代碼repo：https://github.com/geekan/MetaGPT
論文：https://arxiv.org/pdf/2308.00352.pdf核心模塊
MetaGPT論文中給出的架構圖如上所示。
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 = [] # 對應的action/ stage
self._role_id = str(self._setting)
self._rc = RoleContext()在Role中，又有兩個主要組件：RoleSetting和RoleContext。其中RoleSetting存儲了角色的名字、目標等信息。RoleContext則包含了運行中的狀態信息，如記憶，需要執行的動作等。
class RoleSetting(BaseModel):
"""Role Settings"""
name: str
profile: str
goal: str
constraints: str
desc: str
class RoleContext(BaseModel):
"""Role Runtime Context"""
# 維護運行過程中的一些狀態信息
env: 'Environment' = Field(default=None) # environment，所有角色共享
memory: Memory = Field(default_factory=Memory) # 記憶信息
state: int = Field(default=0) # 下一時刻要執行的動作
todo: Action = Field(default=None)
watch: set[Type[Action]] = Field(default_factory=set) # 關注/訂閱的信息
news: list[Type[Message]] = Field(default=[]) # 本輪新增的信息下面簡單介紹一下MetaGPT的流程。
首先，在environment.run()函數中，會依次調用每個角色的role.run()
async def run(self, k=1):
"""處理一次所有信息的運行
Process all Role runs at once
"""
for _ in range(k):
futures = []
# 執行
for role in self.roles.values():
future = role.run()
futures.append(future)
await asyncio.gather(*futures)在role.run()函數中，與前面所述流程類似分為三步：a) _observe()函數觀察環境、更新memory。b) _react()函數思考并執行動作。c) _publish_message()函數將執行結果更新到環境中。
async def run(self, message=None):
"""Observe, and think and act based on the results of the observation"""
if message:
...
# 觀察環境、更新memory
elif not await self._observe():
# If there is no new information, suspend and wait
return
# 執行動作
rsp = await self._react()
# 把結果傳給environment
self._publish_message(rsp)
return rsp_react()函數可以進一步分為_think()和_act()兩個階段。_think()負責決定SOP的狀態轉移，_act()通過調用action.run()完成具體執行。兩個階段可能都需要調用LLM。
自定義擴展
最后在作者給出的examples中，用戶如果想要實現一個自定義場景，一般情況下只需要實現自己的Action和Role類及對應的關鍵函數即可。
相關資料
代碼repo：https://github.com/OpenBMB/AgentVerse
論文：https://arxiv.org/pdf/2308.10848.pdf除此之外，清華大學的開源框架AgentVerse也提供了一個基于LLMs來搭建多智能體交互的框架。
核心模塊
論文中，提供了如上圖所示的流程圖，它的設計流程模擬了人類的決策過程，包含了四個階段，分別是：專家招聘、協作決策、行動執行和評估。
根據包括軟件開發，咨詢和游戲的幾個不同領域的案例表明，這種多智能體合作的方式優于單一智能體，并且優勢明顯。
AgentVerse 在代碼實現上，包含了兩個基礎元素agent和environment， agent不用過多介紹，他的工作行為依賴特定的prompt以及配合大模型llm進行結果生成。
這里的environment就是代表著任務，在environment中通過定義agents以及rules等信息，來確保多智能體的交互流程能夠按照上述進行。
值得注意的是，environment 的rules中包含了5個組建，這些組件分別是： 描述器，順序，選擇器，更新器，以及可見性。他們的作用具體如下，（參考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)對應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)核心流程
用戶初始化，并調用 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()
# 循環執行
while not self.environment.is_done():
asyncio.run(self.environment.step())相關environment流程邏輯如下：
async def step(self) -> List[Message]:
"""Run one step of the environment"""
# 選擇下一時刻的行動的agent(s)
agent_ids = self.rule.get_next_agent_idx(self)
# 環境描述信息（每個agents不一定一樣）
env_descriptions = self.rule.get_env_description(self)
# agent行動，返回結果
messages = await asyncio.gather(
*[self.agents[i].astep(env_descriptions[i]) for i in agent_ids]
)
# 選擇過濾message信息
selected_messages = self.rule.select_message(self, messages)
# 更新mmory
self.rule.update_memory(self)
# 更新agent之間的可見性
self.rule.update_visible_agents(self)
self.cnt_turn += 1
return selected_messages相關 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:
# 調用LLM
response = self.llm.generate_response(prompt)
# 解析結果
parsed_response = self.output_parser.parse(response)
if isinstance(parsed_response, AgentAction):
# 調用工具
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拿軟件開發的任務舉例子，用戶需要提供一個json配置文件，包含environment信息所需的信息，并且定義若干個不同的agent，如code_writer，code_reviwer，unit_test_generator，以及他們之間溝通時候的rule，最終初始化environment， agents以及agentverse入口方法，并調用 agentVerse.run()即可。
相關資料
代碼repo：https://github.com/aiwaves-cn/agents
論文：https://arxiv.org/pdf/2309.07870.pdf最后，波形智能聯合浙大和蘇黎世聯邦理工大學，提供了另外一種支持mutli-agent的設計方案。
核心模塊
該方案圍繞SOP模塊來確定agent之間交互的推進方式，通過SOP來管理狀態的變換，并將相關狀態信息變換記錄到環境中，以便不同的agent進行各自任務的推進。 整體流程如下圖所示：
Agents有三個核心模塊，除了上述SOP，還有Agent以及Environment。
（參考：https://mp.weixin.qq.com/s/toblMJJkpFKtv0dfJFfHKA）
核心流程
入口代碼如下所示，用戶通過配置文件來初始化agent，environment以及sop，并通過如下入口程序開始agent交互流程。
def run(agents,sop,environment):
while True:
# 更新狀態，決定下一個行動的agent
current_state,current_agent= sop.next(environment,agents)
if sop.finished:
os.environ.clear()
break
# agent執行
action = current_agent.step(current_state)
# 更新memory
memory = process(action)
environment.update_memory(memory,current_state)sop的流程方法如下，通過調用transit方法進行狀態轉移的操作，以及用route進行agent的選擇。
def next(self, environment, agents):
"""
Determine the next state and the agent that needs action based on the current situation
"""
# 一些初始化流程：獲取記憶、相關信息等
...
# 下一時刻的狀態
next_state = self.transit(
chat_history=environment.shared_memory["long_term_memory"][
environment.current_chat_history_idx :
],
relevant_history=relevant_history,
environment=environment,
)
# 如果進入終止節點，則直接終止
if next_state.name == self.finish_state_name:
self.finished = True
return None, None
# 更新狀態
self.current_state = next_state
# 決定下一時刻要行動的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的執行方法如下，主要是根據當前環境信息，來進行prompt生成，并調用llm進行生成
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
# 先根據當前環境更新信息
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()
# 調用LLM
response = current_LLM.get_response(
chat_history, system_prompt, last_prompt, stream=True
)
return response,res_dict最后，agent執行結束后，environment會更新本輪環境信息用于下一輪
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):
"""
更新環境的memory信息
"""
MAX_CHAT_HISTORY = eval(os.environ["MAX_CHAT_HISTORY"])
self.shared_memory["long_term_memory"].append(memory)
current_embedding = get_embedding(memory.content)
# 對過去幾輪信息作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 除了獨特的SOP系統以外，相對于其他框架不同的點還有一個重要功能是支持人類使用者扮演multi-agent系統中的一個或多個智能體的功能，可以方便地支持各種人 – 智能體交互的應用場景，如人和智能體一起玩游戲、辯論等。
最后，我們針對上述幾個市面上比較主流的agents框架進行簡單的總結與比較。部分相關特性如下所示：
| 框架名稱 | MetaGPT | AgentVerse | Agents |
| Config的詳細性 | 少 | 中 | 多 |
| 狀態的切換 | 基于LLM與SOP | 無/迭代達到最大輪數 | 基于LLM與SOP |
| Agent執行順序 | 順序 | 基于規則 | 順序/LLM決定/隨機 |
| 長短期記憶 | 有 | 有 | 有/做了額外處理 |
| 工具調用 | 少 | 少/單個 | 少/不是由LLM決定的 |
| Agent差異性體現（即對當前Agent，其他Agent是否有區別） | 有 | 有 | 無 |
綜上，multi-agent框架的核心還是在agent之間的交互與狀態的轉移。而LLM對工具的調用往往只是作為其中一個子模塊，且大部分agent都只調用單個tool。因此，在multi-agent的場景中，使用多個單tool的agent還是一個多tool的agent，也是一個值得考慮的問題。
本文章轉載微信公眾號@青稞AI
