2023-03-11-AutoGen
使用【autoGen+chainlit+deepSeek】實現【多角色、多用戶、多智能體對話系統】
1-核心思路
- 01）技術要點：autoGen+chainlit+deepSeek
- 02）什么是autoGen->autogen是微軟旗下的多智能體的框架
- 03）什么是chainlit->模型前端展示框架
- 04）要做什么？->快速實現多角色、多用戶、多智能體對話系統
- 05）什么是RoundRobinGroupChat？->反復調用
- 06）單智能體-nurses_station_ai ->進行大模型問題咨詢
- 07）智能體與FunctionCall-> 大模型調用FunctionCall
- 08）多智能體自動選擇-> Selector Group Chat
- 09）為什么設置TextMentionTermination->終止符
- 10）什么是AutoGen Studio工作流UI->上面都是python代碼實現，有沒有拖拖拽拽就可以的UI編排
- 11）最終目標->3分鐘復刻Manus智能體！AutoGen+MCP Server+Cline構建最強AI智能體https://www.bilibili.com/video/BV119RHYmEPF
2-參考網址
- AutoGen入門——快速實現多角色、多用戶、多智能體對話系統
- autogen-Github倉庫地址
- 個人代碼實現倉庫地址
3-上手實操
1-安裝依賴
# 版本說明
- python，3.11
- AutoGen，0.4.2
- chainlit，2.0.2
- 大模型，deepseek# 安裝依賴-autogen
pip install -U "autogen-agentchat" "autogen-ext[openai]"# 安裝依賴-UI交互界面
pip install chainlit# 運行腳本
chainlit run .\a_01_nurses_station_ai.py -w
2-a_01_nurses_station_ai內容如下
import chainlit as cl
from autogen_agentchat.agents import AssistantAgent
from autogen_agentchat.teams import RoundRobinGroupChat
from autogen_ext.models.openai import OpenAIChatCompletionClient@cl.on_chat_start
async def main():await cl.Message(content="您好，這里是超級無敵大醫院，有什么可以幫您？").send()async def run_team(query: str):model_client = OpenAIChatCompletionClient(model="deepseek-chat", base_url="https://api.deepseek.com",api_key="sk-6d775065fa8c405caebed674c125e67e",model_info={"vision": False,"function_calling": False,"json_output": True,"family": "unknown",}, )assistant_agent = AssistantAgent("assistant", model_client=model_client,system_message="你是一所口腔醫院的導診臺機器人，負責解答用戶的掛號問題，用戶描述癥狀需求，你回答應該掛的科室。""在本醫院中有以下科室：牙體牙髓科、口腔修復科、口腔外科、口腔種植科、兒童口腔專科。""如果用戶的問題與掛號咨詢不符合，回答：“您的描述與癥狀無關，暫不支持”")team = RoundRobinGroupChat(participants=[assistant_agent], max_turns=1)response_stream = team.run_stream(task=query)async for msg in response_stream:if hasattr(msg, "source") and msg.source != "user" and hasattr(msg, "content"):msg = cl.Message(content=msg.content, author="Agent Team")await msg.send()@cl.on_message
async def main(message: cl.Message):await run_team(message.content)3-智能體與FunctionCall
import chainlit as cl
from autogen_agentchat.agents import AssistantAgent
from autogen_agentchat.teams import RoundRobinGroupChat
from autogen_ext.models.openai import OpenAIChatCompletionClient@cl.password_auth_callback
def auth_callback(username: str, password: str):if (username, password) == ("admin", "admin"):return cl.User(identifier="admin", metadata={"role": "admin", "provider": "credentials"})elif (username, password) == ("puhaiyang", "123456"):return cl.User(identifier="puhaiyang", metadata={"role": "admin", "provider": "credentials"})else:return None@cl.on_chat_start
async def main():await cl.Message(content="您好，這里是牙體牙髓科，您牙齒哪里不適？").send()async def x_p_search(tooth_position: str) -> str:"""Find information on the web"""app_user = cl.user_session.get("user")print(f"模擬查詢{app_user.identifier}的{tooth_position}牙片數據")if tooth_position == "46":return "牙根尖處有陰影，疑似感染，需要進一步分析診斷"else:return f"{tooth_position}無影像"async def run_team(query: str):model_client = OpenAIChatCompletionClient(model="deepseek-chat", base_url="https://api.deepseek.com",api_key="sk-6d775065fa8c405caebed674c125e67e",model_info={"vision": False,"function_calling": True,"json_output": True,"family": "unknown",})assistant_agent = AssistantAgent("assistant", model_client=model_client, tools=[x_p_search],system_message="你是一個牙體牙髓科的病情診斷機器人，負責對用戶輸入的癥狀描述分析原因，在分析病因前先詢問出用戶是具體哪一顆牙齒需要治療。""在知道了具體的牙位號后，再調用x_p_search工具進行問題回答，傳入給x_p_search工具的參數需要自動轉為牙位號，如:28""如果用戶的問題與病情咨詢無關，回答：“您的描述與癥狀無關，暫不支持”")team = RoundRobinGroupChat(participants=[assistant_agent], max_turns=1)response_stream = team.run_stream(task=query)async for msg in response_stream:if hasattr(msg, "source") and (msg.type == "ToolCallExecutionEvent" or msg.type == "ToolCallRequestEvent"):# functionCall事件消息不顯示給用戶continueif hasattr(msg, "source") and msg.source != "user" and hasattr(msg, "content"):if msg.content.endswith("無影像"):res = await cl.AskActionMessage(content=f"{msg.content}，是否需要幫您申請拍攝此牙的CT影像？",actions=[cl.Action(name="continue", payload={"value": "申請"}, label="? 申請牙片"),cl.Action(name="cancel", payload={"value": "取消"}, label="? 取消"),],).send()if res and res.get("payload").get("value") == "申請":await cl.Message(content="牙片申請已提交!待審核通過后前往第3影像室進行拍攝。",).send()else:msg = cl.Message(content=msg.content, author="Agent Team")await msg.send()@cl.on_message
async def main(message: cl.Message):await run_team(message.content)- 執行腳本
# 1-因為添加了chainlit認證配置
conda activate AutoGen3_11
chainlit create-secret# 2-配置JWT-創建.env文件
CHAINLIT_AUTH_SECRET="-Nj%Z@l^x0FQcEKYuNBxJ@mYr~2L>_ua6EU$zSsR6M~Fd.ssav5GFANVyR4%S$Zf"# 3-執行腳本
chainlit run .\a_02_endodontics_dentistry_ai.py -w
- 觸發大模型的FunctionCall功能
4-如何解決chainlit報錯
ValueError: You must provide a JWT secret in the environment to use authentication. Run
chainlit create-secretto
generate one.
這個錯誤信息表明，在使用需要身份驗證的功能時，你沒有在環境變量中提供 JWT（JSON Web Token）密鑰。JWT
是一種用于在網絡應用中安全傳輸信息的開放標準，通常用于身份驗證和授權。下面為你詳細分析和解決這個問題：
1. 生成 JWT 密鑰
依據錯誤提示，你可以運行 chainlit create-secret 命令來生成一個 JWT 密鑰。打開命令行終端，激活你的 Python
虛擬環境（AutoGen3_11），然后執行以下命令：
conda activate AutoGen3_11
chainlit create-secret
執行該命令后，會輸出一個隨機生成的 JWT 密鑰，類似如下：
CHAINLIT_AUTH_SECRET="-Nj%Z@l^x0FQcEKYuNBxJ@mYr~2L>_ua6EU$zSsR6M~Fd.ssav5GFANVyR4%S$Zf"
2. 根目錄創建.env文件
將剛才的生成的信息復制進去
CHAINLIT_AUTH_SECRET="-Nj%Z@l^x0FQcEKYuNBxJ@mYr~2L>_ua6EU$zSsR6M~Fd.ssav5GFANVyR4%S$Zf"
5-多智能體自動選擇
核心代碼：創建多個AssistantAgent放入到SelectorGroupChat供大模型根據上線文進行選擇
import chainlit as clfrom autogen_agentchat.agents import AssistantAgent
from autogen_agentchat.conditions import MaxMessageTermination, TextMentionTermination
from autogen_agentchat.teams import SelectorGroupChat
from autogen_ext.models.openai import OpenAIChatCompletionClientmodel_client = OpenAIChatCompletionClient(model="deepseek-chat", base_url="https://api.deepseek.com",api_key="sk-6d775065fa8c405caebed674c125e67e",model_info={"vision": False,"function_calling": True,"json_output": True,"family": "unknown",}, )planning_agent = AssistantAgent("PlanningAgent",description="用于規劃的Agent，當一個任務到達時此Agent是第一個參與者",model_client=model_client,system_message="""你是一個任務規劃智能體。你的工作是將復雜的任務分解為更小的、可管理的子任務。你的團隊成員有3個，分別是：DentalPulpAgent: 牙體牙髓科智能體RestorativeAgent: 牙齒修復科智能體DentalImplantAgent: 牙齒種植科智能體你只計劃和委派任務，而不自己執行它們分配任務時，請使用此格式:1. <agent> : <task>當所有智能體把任務完成后，再總結結果以"TERMINATE"結束。 """)dental_pulp_agent = AssistantAgent("DentalPulpAgent",description="牙體牙髓科智能體",model_client=model_client,system_message="""你是一個口腔醫院的牙體牙髓科智能體。你可以解答關于牙體牙髓科中患者提出的問題，你的解答非常專業，且可靠。""")restorative_agent = AssistantAgent("RestorativeAgent",description="牙齒修復科智能體",model_client=model_client,system_message="""你是一個口腔醫院的牙齒修復科智能體。你可以解答關于牙齒修復中患者提出的問題，比如牙冠、烤瓷牙、嵌體修復等。你的解答非常專業，且可靠。""")dental_implant_agent = AssistantAgent("DentalImplantAgent",description="牙齒種植科智能體",model_client=model_client,system_message="""你是一個口腔醫院的牙齒種植科的智能體。你可以解答關于牙齒種植科中患者提出的問題，你的解答非常專業，且可靠。""")@cl.on_chat_start
async def main():await cl.Message(content="您好，這里是口腔醫院專家團隊，有什么可以幫您？").send()async def run_team(query: str):text_mention_termination = TextMentionTermination("TERMINATE")max_messages_termination = MaxMessageTermination(max_messages=25)termination = text_mention_termination | max_messages_terminationteam = SelectorGroupChat([planning_agent, dental_pulp_agent, restorative_agent, dental_implant_agent],model_client=model_client,termination_condition=termination,)response_stream = team.run_stream(task=query)async for msg in response_stream:if hasattr(msg, "source") and msg.source != "user" and hasattr(msg, "content"):msg = cl.Message(content=msg.content, author=msg.source)await msg.send()@cl.on_message
async def main(message: cl.Message):await run_team(message.content)- 執行腳本
chainlit run .\a_03_dentistry_selector_ai.py -w
- 測試問題
什么是烤瓷牙？
什么是根管治療？
5-AutoGen Studio工作流UI
比較好的是AutoGen中也提供了與Dify類似UI界面操作的方式，即：AutoGen Studio。
- 安裝命令
# 安裝autogenstudio
pip install -U autogenstudio# 運行腳本-指定一個文件夾進行數據存放
autogenstudio ui --port 8081 --appdir autogenstuido_test# 程序運行打印
2025-03-12 01:07:58.662 | INFO | autogenstudio.web.app:lifespan:35 - Application startup complete. Navigate to http://127.0.0.1:8081
)
)
會出現的問題)
