Magentic-ui项目相关整理

src/magentic_ui/
├── teams/
│   ├── orchestrator/
│   │   ├── _group_chat.py      # GroupChat 类定义，管理参与者的状态
│   │   ├── _orchestrator.py    # Orchestrator 类定义，维护对话的状态
│   │   └── orchestrator_config.py    # 相关的配置 
│   └── roundrobin_orchestrator.py    # 轮询状态的管理
├── agents/
│   ├── file_surfer/
│   │   └── _file_surfer.py
│   ├── web_surfer/
│   │   └── _web_surfer.py
│   └── users/
│       ├── _dummy_user_proxy.py
│       └── _metadata_user_proxy.py
├── backend/
│   ├── teammanager/
│       └── _teammanager.py
└── task_team.py

# src\magentic_ui\task_team.py
# 用户代理的创建
user_proxy: DummyUserProxy | MetadataUserProxy | UserProxyAgentif magentic_ui_config.user_proxy_type == "dummy":user_proxy = DummyUserProxy(name="user_proxy")elif magentic_ui_config.user_proxy_type == "metadata":assert (magentic_ui_config.task is not None), "Task must be provided for metadata user proxy"assert (magentic_ui_config.hints is not None), "Hints must be provided for metadata user proxy"assert (magentic_ui_config.answer is not None), "Answer must be provided for metadata user proxy"user_proxy = MetadataUserProxy(name="user_proxy",description="Metadata User Proxy Agent",task=magentic_ui_config.task,helpful_task_hints=magentic_ui_config.hints,task_answer=magentic_ui_config.answer,model_client=model_client_orch,)else:user_proxy_input_func = make_agentchat_input_func(input_func)user_proxy = UserProxyAgent(description=USER_PROXY_DESCRIPTION,name="user_proxy",input_func=user_proxy_input_func,)# 轮询代理if websurfer_loop_team:# simplified team of only the web surferteam = RoundRobinGroupChat(participants=[web_surfer, user_proxy],max_turns=10000,)await team.lazy_init()return team# 由LLM智能选择team = GroupChat(participants=[web_surfer, user_proxy, coder_agent, file_surfer],orchestrator_config=orchestrator_config,model_client=model_client_orch,memory_provider=memory_provider,)await team.lazy_init()return team

async def on_messages_stream(self, messages: Sequence[BaseChatMessage], cancellation_token: CancellationToken
) -> AsyncGenerator[BaseAgentEvent | BaseChatMessage | Response, None]:"""处理输入消息并生成响应流这是核心方法，处理所有消息逻辑"""# 1. 严格模式下的提示重写if (self.how_helpful == "strict"and self.helpful_task_hintsand self.helpful_task_hints != "Helpful hints are not available for this task."and self.rewritten_helpful_task_hints is None):self.rewritten_helpful_task_hints = await self._rewrite_helpful_hints(cancellation_token)# 2. 消息处理chat_messages = thread_to_context(list(messages),agent_name=self.name,is_multimodal=self._model_client.model_info["vision"],)self._chat_history.extend(chat_messages)# 3. 阶段判断if ("type" in messages[-1].metadataand messages[-1].metadata["type"] == "plan_message"):self.have_encountered_plan_message = Trueself.in_planning_phase = Trueelse:if self.have_encountered_plan_message:self.in_planning_phase = Falseelse:self.in_planning_phase = True# 4. 模型上下文准备await self._model_context.clear()system_message = SystemMessage(content=self._get_system_message())await self._model_context.add_message(system_message)# 5. 添加聊天历史for msg in self._chat_history:await self._model_context.add_message(msg)# 6. 获取token限制的历史记录token_limited_history = await self._model_context.get_messages()# 7. 规划阶段处理if self.in_planning_phase:if self.simulated_user_type in ["co-planning", "co-planning-and-execution"]:if (self.max_co_planning_rounds is Noneor self.current_co_planning_round < self.max_co_planning_rounds):# 生成规划响应result = await self._model_client.create(messages=token_limited_history,cancellation_token=cancellation_token,)yield Response(chat_message=TextMessage(content=result.content,source=self.name,metadata={"co_planning_round": str(self.current_co_planning_round),"user_plan_reply": "llm","helpful_task_hints": self.rewritten_helpful_task_hintsif self.rewritten_helpful_task_hintselse self.helpful_task_hints,},),inner_messages=[],)self.current_co_planning_round += 1else:# 达到最大规划轮次yield Response(chat_message=TextMessage(content="accept",source=self.name,metadata={"co_planning_round": str(self.current_co_planning_round),"user_plan_reply": "accept",},),inner_messages=[],)else:# 非协作规划模式yield Response(chat_message=TextMessage(content="accept",source=self.name,metadata={"co_planning_round": str(self.current_co_planning_round),"user_plan_reply": "accept",},),inner_messages=[],)# 8. 执行阶段处理else:if self.simulated_user_type in ["co-execution", "co-planning-and-execution"]:if (self.max_co_execution_rounds is Noneor self.current_co_execution_round < self.max_co_execution_rounds):# 生成执行响应result = await self._model_client.create(messages=token_limited_history,cancellation_token=cancellation_token,)yield Response(chat_message=TextMessage(content=result.content,source=self.name,metadata={"user_execution_reply": "llm"},),inner_messages=[],)self.current_co_execution_round += 1else:# 达到最大执行轮次yield Response(chat_message=TextMessage(content="I don't know, you figure it out, don't ask me again.",source=self.name,),)else:# 非协作执行模式yield Response(chat_message=TextMessage(content="I don't know, you figure it out, don't ask me again.",source=self.name,metadata={"user_execution_reply": "idk"},),)

 @rpcasync def handle_start(self, message: GroupChatStart, ctx: MessageContext) -> None:
# @rpc 是远程过程调用，运行函数被远程调用，就像调用本地函数一样
# message 来自于外部的调用.venv\Lib\site-packages\autogen_agentchat\teams\_group_chat\_base_group_chat.py 
文件中的 run_stream 方法，就是负责将初始值（初始消息）通过 GroupChatStart 消息传递
给群聊管理器（GroupChatManager）的地方。具体流程如下：'''
对第一点的self.team的解释，实际上self.team就是代指GroupChat
路径：src\magentic_ui\teams\orchestrator\_group_chat.py下的run_stream谁去调用？
不会直接 new 一个 GroupChat 然后直接 run_stream，而是通过 TeamManager 统一管理
GroupChat.run_stream 的实际调用，主要是通过 TeamManager 的 self.team.run_stream 
实现的，这就是 “间接调用”
(本文件的)14，15，16行在src\magentic_ui\backend\teammanager\teammanager.py下的run_stream 调用
的313到315行代码中async for message in self.team.run_stream(task=task, cancellation_token=cancellation_token):
其中的self.team.run_stream也就代指了GroupChat.run_stream
src\magentic_ui\backend\teammanager\teammanager.py 的 run_stream 由
src\magentic_ui\backend\web\managers\connection.py 下start_stream方法的run_stream调用
'''
1.调用 self.team.run_stream(task=...)，其中 task 可以是字符串、消息对象、
消息列表，或者为 None。2.run_stream 方法会把 task 转换成 messages，然后封装到 GroupChatStart(
messages=messages) 这个消息对象里。（路径：.venv\Lib\site-packages\autogen_agentchat\teams\_group_chat\_base_group_chat.py）if task is None:# 如果 task 是 None，则 messages 也是 Nonepasselif isinstance(task, str):# 如果 task 是字符串，就会被包装成一个 TextMessage，放到 messages 列表里messages = [TextMessage(content=task, source="user")]elif isinstance(task, BaseChatMessage):# 如果 task 是单个消息对象，也会被放到 messages 列表里。messages = [task]elif isinstance(task, list):# 如果 task 是消息对象列表，则直接赋值给 messages。...# 也就是把 messages 作为参数，封装进 GroupChatStart 消息对象里  499行await self._runtime.send_message(GroupChatStart(messages=messages),recipient=AgentId(type=self._group_chat_manager_topic_type, key=self._team_id),cancellation_token=cancellation_token,)3.通过 self._runtime.send_message(...)，把 GroupChatStart 
消息发送给群聊管理器（GroupChatManager）。
# .venv/Lib/site-packages/autogen_agentchat/teams/_group_chat/_base_group_chat.py 中的run_stream方法await self._runtime.send_message(GroupChatStart(messages=messages),recipient=AgentId(type=self._group_chat_manager_topic_type, key=self._team_id),cancellation_token=cancellation_token,)
# recipient=AgentId(type=self._group_chat_manager_topic_type, key=self._team_id) 
# 明确指定了接收者就是群聊管理器， _group_chat_manager_topic_type 就是Orchestrator这个群聊管理器实例4.群聊管理器收到 GroupChatStart 后，会调用 handle_start 方法，正式启动群聊流程。
# 对于 GroupChatStart 消息，自动调用handle_start方法，message参数就是刚收到的 GroupChatStart消息