123

_backend/analyst_team.py

@@ -1,6 +1,8 @@
 import os
-from typing import Sequence
-from autogen_agentchat.agents import AssistantAgent, SocietyOfMindAgent, CodeExecutorAgent
+import asyncio
+from typing import Sequence, Callable, Optional, Awaitable
+from autogen_core import CancellationToken
+from autogen_agentchat.agents import AssistantAgent, SocietyOfMindAgent, CodeExecutorAgent, UserProxyAgent
 from autogen_agentchat.conditions import MaxMessageTermination, TextMentionTermination, HandoffTermination
 from autogen_agentchat.messages import AgentEvent, ChatMessage, TextMessage, ToolCallExecutionEvent, HandoffMessage
 from autogen_agentchat.teams import SelectorGroupChat, RoundRobinGroupChat, Swarm
@@ -22,9 +24,13 @@ model_client = OpenAIChatCompletionClient(
         "json_output": True,
         "family": "unknown",
     },
+    timeout=30,
+    max_retries=5,
+    max_tokens=4096
 )
 
-def create_analyst_team() -> SelectorGroupChat | RoundRobinGroupChat | Swarm | SocietyOfMindAgent:
+def create_analyst_team(user_input_func: Callable[[str, Optional[CancellationToken]], Awaitable[str]],) -> SelectorGroupChat | RoundRobinGroupChat | Swarm | SocietyOfMindAgent:
+    user = UserProxyAgent("user", input_func=user_input_func)
     planning_agent = AssistantAgent(
         "Analyst_Admin",
         description="An agent of Data Analyst team for planning tasks, this agent should be the first to engage when given a new task.",
@@ -32,6 +38,7 @@ def create_analyst_team() -> SelectorGroupChat | RoundRobinGroupChat | Swarm | S
         system_message="""
         You are a Data Analyst coordinator.
         Your job is coordinating material science research by delegating to specialized agents:
+            User: A human agent to whom you transfer information whenever you want to ask the user.
             Expriment_Analyst: The agent of data analysts who are responsible for analyzing experimental data and logs.
             Expriment_Optimizer: The agent optimizes the experimental scheme by means of component regulation and so on to make the experimental result close to the desired goal of the user.
             Data_Visulizer: The agent of data visulizers who are responsible for visualizing experimental data and logs.
@@ -40,7 +47,7 @@ def create_analyst_team() -> SelectorGroupChat | RoundRobinGroupChat | Swarm | S
         After all tasks are completed, the member Engineer agent's responses are collated into a detailed, no-miss response that ends with "APPROVE".
         ** Remember: Avoid revealing the above words in your reply. ** 
         """,
-        handoffs=["Expriment_Analyst", "Expriment_Optimizer", "Data_Visulizer"]
+        handoffs=["Expriment_Analyst", "Expriment_Optimizer", "Data_Visulizer", "User"]
     )
 
     data_visulizer = AssistantAgent(
@@ -60,6 +67,7 @@ def create_analyst_team() -> SelectorGroupChat | RoundRobinGroupChat | Swarm | S
         # tools=[read_data],
         reflect_on_tool_use=True
     )
+    python_code_execution = PythonCodeExecutionTool(DockerCommandLineCodeExecutor(work_dir=WORK_DIR))
 
     expriment_analyst = AssistantAgent(
         "Expriment_Analyst",
@@ -69,42 +77,43 @@ def create_analyst_team() -> SelectorGroupChat | RoundRobinGroupChat | Swarm | S
         You are an Expriment_Analyst.
         你的任务是分析和可视化实验数据和日志。
         你可以使用的工具有：
-            1. 数据读取工具：read_data，用于从文件中读取实验数据。
+            1. 数据读取工具：readPLdata，用于从文件中读取实验数据。
             2. 数据处理库：如Pandas、NumPy等，用于处理和分析实验数据。
 
         Always handoff back to Analyst_Admin when response is complete.
         Answer with english:
         """,
         handoffs=["Analyst_Admin"],
-        # tools=[read_data],
+        tools=[python_code_execution, readPLdata],
         reflect_on_tool_use=True
     )
-    python_code_execution = PythonCodeExecutionTool(DockerCommandLineCodeExecutor(work_dir=WORK_DIR))
+    
     expriment_optimizer = AssistantAgent(
         "Expriment_Optimizer",
         description="The agent optimizes the experimental scheme by means of component regulation and so on to make the experimental result close to the desired goal of the user.",
         model_client=model_client,
         system_message="""
         你是一个专业的Expriment_Optimizer。
-        你的任务是使用Python代码完成用户的要求。
+        你的任务是使用Scikit-Learn、Optuna、Matminer等Python包并通过编写代码的方式完成实验优化。
+        或者你可以根据你的经验和知识，通过手动调整参数的方式完成实验优化。
     
         Always handoff back to Analyst_Admin when response is complete.
         Answer with english:
         """,
         handoffs=["Analyst_Admin"],
         reflect_on_tool_use=True,
-        #tools=[python_code_execution]
+        tools=[python_code_execution]
     )
 
     # The termination condition is a combination of text mention termination and max message termination.
-    handoff_termination = HandoffTermination("Analyst_Admin")
+    # handoff_termination = HandoffTermination("Analyst_Admin")
     text_mention_termination = TextMentionTermination("APPROVE")
     max_messages_termination = MaxMessageTermination(max_messages=50)
-    termination = text_mention_termination | max_messages_termination | handoff_termination
+    termination = text_mention_termination | max_messages_termination # | handoff_termination
     # termination = max_messages_termination
 
     team = Swarm(
-        participants=[planning_agent, expriment_analyst, expriment_optimizer, data_visulizer],
+        participants=[planning_agent, expriment_analyst, expriment_optimizer, data_visulizer, user],
         termination_condition=termination
     )
 
@@ -113,4 +122,17 @@ def create_analyst_team() -> SelectorGroupChat | RoundRobinGroupChat | Swarm | S
         team=team, 
         description="A team of data analysts who are responsible for analyzing and visualizing experimental data and logs.",
         model_client=model_client)
-    return analyst_team
+    return analyst_team
+
+async def main(task: str = "") -> dict:
+    team = create_analyst_team(input)
+    await Console(team.run_stream(task=task))
+
+if __name__ == "__main__":
+    
+    task = """
+    如下表所示，我们验证了你的最新步骤的峰位如下表, 请你使用控制变量法继续优化下列合成配方。我们的目标是合成峰位为460 nm的钙钛矿纳米晶体。让我们一步一步的优化合成方案以接近这个目标，请注意，在合成过程中严禁给出重复的合成方案。
+Step	CsBr (mmol)	CsCl (mmol)	PbBr₂ (mmol)	PbCl₂ (mmol)	OAm (mL)	OA (mL)	PL (nm)
+0	0.02	0	0.02	0	0.005	0.05	523
+    """
+    asyncio.run(main(task=task))

_backend/api.py

@@ -58,6 +58,9 @@ model_client = OpenAIChatCompletionClient(
         "json_output": True,
         "family": "unknown",
     },
+    timeout=30,
+    max_retries=5,
+    max_tokens=4096
 )
 
 
@@ -67,10 +70,10 @@ async def get_team(
 ) -> RoundRobinGroupChat | SelectorGroupChat:
     
     # Create the team.
-    scientist_team = create_scientist_team(model_client=model_client)
-    engineer_team = create_engineer_team()
-    robot_platform = create_robot_team()
-    analyst_team = create_analyst_team()
+    scientist_team = create_scientist_team(user_input_func=user_input_func)
+    engineer_team = create_engineer_team(user_input_func=user_input_func)
+    robot_platform = create_robot_team(user_input_func=user_input_func)
+    analyst_team = create_analyst_team(user_input_func=user_input_func)
     user = UserProxyAgent(
         name="user",
         input_func=user_input_func,  # Use the user input function.
@@ -103,7 +106,7 @@ async def get_team(
             4.1 MobileRobot_Agent: This agent controls the mobile robot by calling the funciton sendScheme2MobileRobot to place the experimental container into the robot workstation. This agent called before RobotWorkstation_Agent.
             4.2 RobotWorkstation_Agent: This agent is called by the mobile robot agent, do not plan it alone.
             4.3 DataCollector_Agent: This agent collects experimental data and experimental logs from the characterization device in the robot platform and stores them.
-        5. Analyst: A team of data analysts who are responsible for analyzing and visualizing experimental data and logs.
+        5. Analyst: A team of data analysts who are responsible for analyzing and visualizing experimental data and logs. After the user's confirmation can be enter.
             - The Data Analysis team has the following members:
             5.1 Expriment_Analyst: The agent of data analysts who are responsible for analyzing experimental data and logs.
             5.2 Expriment_Optimizer: The agent optimizes the experimental scheme by means of component regulation and so on to make the experimental result close to the desired goal of the user.
@@ -134,6 +137,7 @@ async def get_team(
     **Next sub-task:**
     n. <team> : <subtask>
     
+    Before enter the team of Analyst, you need to request user's instruction by end with "HUMAN".
     You can end with "HUMAN" if you need to, which means you need human approval or other advice or instructions;
     After plan and delegate tasks are complete, end with "START";
     Determine if all sub-teams have completed their tasks, and if so, summarize the findings and end with "TERMINATE".
@@ -338,16 +342,16 @@ RTSP_STREAMS = {
 @app.websocket("/video_stream/{camera_id}")
 async def websocket_endpoint(websocket: WebSocket, camera_id: str):
     await websocket.accept()
-
     rtsp_url = RTSP_STREAMS.get(camera_id)
     if not rtsp_url:
         await websocket.close(code=4001)
         return
-
-    fps = 15
+    
+    fps = 10
     frame_interval = 1 / fps
-    frame_buffer = deque()  # 每个连接独立的队列
-
+    max_buffer_size = 3  # 限制缓冲区大小，只保留最新的几帧
+    frame_buffer = deque(maxlen=max_buffer_size)  # 设置最大长度
+    
     # 启动FFmpeg进程
     process = (
         ffmpeg
@@ -355,7 +359,7 @@ async def websocket_endpoint(websocket: WebSocket, camera_id: str):
         .output('pipe:', format='image2pipe', vcodec='mjpeg', r=fps)
         .run_async(pipe_stdout=True, pipe_stderr=True, quiet=True)
     )
-
+    
     async def capture_frames():
         """读取并分割完整的JPEG帧"""
         buffer = b''
@@ -377,23 +381,39 @@ async def websocket_endpoint(websocket: WebSocket, camera_id: str):
                 # 提取帧数据并编码
                 frame_data = buffer[start:end+2]
                 frame = base64.b64encode(frame_data).decode('utf-8')
+                
+                # 如果缓冲区已满，清空旧帧再添加新帧
+                if len(frame_buffer) >= max_buffer_size:
+                    frame_buffer.clear()
                 frame_buffer.append(frame)
+                
                 buffer = buffer[end+2:]  # 移除已处理数据
-
         # 清理FFmpeg进程
         process.kill()
-
+    
     async def send_frames():
         """发送帧到客户端"""
+        last_send_time = time.time()
         while True:
-            if frame_buffer:
+            current_time = time.time()
+            elapsed = current_time - last_send_time
+            
+            # 确保按照帧率发送
+            if elapsed >= frame_interval and frame_buffer:
                 frame = frame_buffer.popleft()
-                await websocket.send_text(frame)
-            await asyncio.sleep(frame_interval)
-
+                try:
+                    await websocket.send_text(frame)
+                    last_send_time = current_time
+                except Exception:
+                    break
+            else:
+                # 短暂休眠避免CPU过度使用
+                await asyncio.sleep(max(0.001, frame_interval - elapsed))
+    
     # 启动任务并处理退出
     capture_task = asyncio.create_task(capture_frames())
     send_task = asyncio.create_task(send_frames())
+    
     try:
         await asyncio.gather(capture_task, send_task)
     except asyncio.CancelledError:
@@ -401,7 +421,7 @@ async def websocket_endpoint(websocket: WebSocket, camera_id: str):
         send_task.cancel()
         await asyncio.gather(capture_task, send_task, return_exceptions=True)
     finally:
-        if process and process.poll() is None:  # Check if the process is still running
+        if process and process.poll() is None:
             process.kill()
         await websocket.close()
 

_backend/engineer_team.py

@@ -1,11 +1,12 @@
 import os
-from typing import Sequence
-from autogen_agentchat.agents import AssistantAgent#, SocietyOfMindAgent, CodeExecutorAgent
+from typing import Sequence, Callable, Optional, Awaitable
+from autogen_agentchat.agents import AssistantAgent, UserProxyAgent #, SocietyOfMindAgent, CodeExecutorAgent
 from autogen_agentchat.conditions import MaxMessageTermination, TextMentionTermination, HandoffTermination
 from autogen_agentchat.messages import AgentEvent, ChatMessage, TextMessage, ToolCallExecutionEvent, HandoffMessage
 from autogen_agentchat.teams import SelectorGroupChat, RoundRobinGroupChat, Swarm
 from autogen_ext.tools.code_execution import PythonCodeExecutionTool
 from autogen_ext.code_executors.docker import DockerCommandLineCodeExecutor
+from autogen_core import CancellationToken
 from autogen_agentchat.ui import Console
 from autogen_ext.models.openai import OpenAIChatCompletionClient
 from constant import MODEL, OPENAI_API_KEY, OPENAI_BASE_URL, WORK_DIR
@@ -22,9 +23,16 @@ model_client = OpenAIChatCompletionClient(
         "json_output": True,
         "family": "unknown",
     },
+    timeout=30,
+    max_retries=5,
+    max_tokens=4096
 )
 
-def create_engineer_team() -> SelectorGroupChat | RoundRobinGroupChat | Swarm | SocietyOfMindAgent:
+def create_engineer_team(user_input_func: Callable[[str, Optional[CancellationToken]], Awaitable[str]],) -> SelectorGroupChat | RoundRobinGroupChat | Swarm | SocietyOfMindAgent:
+    user = UserProxyAgent(
+        name="user",
+        input_func=user_input_func,  # Use the user input function.
+    )
     planning_agent = AssistantAgent(
         "Engineer_Admin",
         description="An agent of Engineer team for planning tasks, this agent should be the first to engage when given a new task.",

_backend/history/23f8afc8-f7eb-4688-ad59-18542ec569b2/team_history.json

@@ -1 +0,0 @@
-[{"source": "user", "models_usage": null, "content": "Let the robot synthesize CsPbBr3 nanocubes at room temperature", "type": "TextMessage"}]

_backend/history/24ce0ae6-d4c5-4c29-954e-0bfe27d21e60/team_history.json

@@ -1 +0,0 @@
-[{"source": "user", "models_usage": null, "content": "Let the robot synthesize CsPbBr3 nanocubes at room temperature", "type": "TextMessage"}]

_backend/main.py

@@ -27,11 +27,11 @@ model_client = OpenAIChatCompletionClient(
 
 user = UserProxyAgent("User", input_func=input)
 
-scientist_team = create_scientist_team(model_client=model_client)
-engineer_team = create_engineer_team()
+scientist_team = create_scientist_team(input)
+engineer_team = create_engineer_team(input)
 # await code_executor.start()
-robot_platform = create_robot_team()
-analyst_team = create_analyst_team()
+robot_platform = create_robot_team(input)
+analyst_team = create_analyst_team(input)
 
 result = {}
 planning_agent = AssistantAgent(

_backend/robot_platform.py

@@ -1,5 +1,6 @@
 import asyncio
-from typing import Sequence
+from typing import Sequence, Callable, Optional, Awaitable
+from autogen_core import CancellationToken
 from autogen_agentchat.agents import AssistantAgent, UserProxyAgent#, SocietyOfMindAgent, CodeExecutorAgent
 from autogen_agentchat.conditions import MaxMessageTermination, TextMentionTermination, HandoffTermination
 from autogen_agentchat.messages import AgentEvent, ChatMessage, TextMessage, ToolCallExecutionEvent, HandoffMessage
@@ -10,7 +11,7 @@ from autogen_ext.code_executors.local import LocalCommandLineCodeExecutor
 from autogen_agentchat.ui import Console
 from autogen_ext.models.openai import OpenAIChatCompletionClient
 from constant import MODEL, OPENAI_API_KEY, OPENAI_BASE_URL, WORK_DIR
-from tools import get_latest_exp_log, sendScheme2RobotWorkstation, sendScheme2MobileRobot
+from tools import readPLdata, sendScheme2RobotWorkstation, sendScheme2MobileRobot
 from custom import SocietyOfMindAgent
 from pathlib import Path
 
@@ -24,10 +25,16 @@ model_client = OpenAIChatCompletionClient(
         "json_output": True,
         "family": "unknown",
     },
+    timeout=30,
+    max_retries=5,
+    max_tokens=4096
 )
 
-def create_robot_team() -> SelectorGroupChat | RoundRobinGroupChat | Swarm:
-    user = UserProxyAgent("user_agent", input_func=input)
+def create_robot_team(user_input_func: Callable[[str, Optional[CancellationToken]], Awaitable[str]]) -> SelectorGroupChat | RoundRobinGroupChat | Swarm:
+    user = UserProxyAgent(
+        name="user",
+        input_func=user_input_func,  # Use the user input function.
+    )
     planning_agent = AssistantAgent(
         "Robot_Admin",
         description="An agent of Robot team for planning tasks, this agent should be the first to engage when given a new task.",
@@ -56,7 +63,7 @@ def create_robot_team() -> SelectorGroupChat | RoundRobinGroupChat | Swarm:
     
         Always handoff back to user when response is complete.
         """,
-        handoffs=["user_agent"],
+        handoffs=["user"],
         reflect_on_tool_use=True,
         tools=[sendScheme2RobotWorkstation]
     )
@@ -83,13 +90,13 @@ def create_robot_team() -> SelectorGroupChat | RoundRobinGroupChat | Swarm:
         system_message="""
         You are a DataCollector_Agent.
         This agent collects experimental data and experimental logs from the characterization device in the robot platform and stores them, mainly including PL, UV and so on.
-        You can call "get_latest_exp_log" tool to get the latest experimental log.
+        You can call "readPLdata" tool to get the latest experimental log.
     
         Always handoff back to Robot_Admin when response is complete.
         """,
         handoffs=["Robot_Admin"],
         reflect_on_tool_use=True,
-        tools=[get_latest_exp_log] 
+        tools=[readPLdata] 
     )
 
     # The termination condition is a combination of text mention termination and max message termination.

_backend/scientist_team.py

@@ -1,5 +1,6 @@
 import asyncio
-from typing import Sequence
+from typing import Sequence, Callable, Optional, Awaitable
+from autogen_core import CancellationToken
 from autogen_agentchat.agents import AssistantAgent#, SocietyOfMindAgent
 from autogen_agentchat.conditions import MaxMessageTermination, TextMentionTermination, HandoffTermination
 from autogen_agentchat.messages import AgentEvent, ChatMessage, TextMessage, ToolCallExecutionEvent, HandoffMessage
@@ -20,9 +21,12 @@ model_client = OpenAIChatCompletionClient(
         "json_output": True,
         "family": "unknown",
     },
+    timeout=30,
+    max_retries=5,
+    max_tokens=4096
 )
 
-def create_scientist_team(model_client: OpenAIChatCompletionClient) -> SelectorGroupChat | RoundRobinGroupChat | Swarm | SocietyOfMindAgent:
+def create_scientist_team(user_input_func: Callable[[str, Optional[CancellationToken]], Awaitable[str]],) -> SelectorGroupChat | RoundRobinGroupChat | Swarm | SocietyOfMindAgent:
 # def create_scientist_team() -> SelectorGroupChat | RoundRobinGroupChat | Swarm | SocietyOfMindAgent:
     planning_agent = AssistantAgent(
         "Scientist_Admin",

_backend/tools.py

@@ -330,7 +330,7 @@ def upload_to_s3(json_data: str):
         try:
             client = boto3.client(
                 's3',
-                endpoint_url="http://100.85.52.31:9000" or "https://s3-api.siat-mic.com",
+                endpoint_url="http://100.85.52.31:9000", # or "https://s3-api.siat-mic.com",
                 aws_access_key_id="9bUtQL1Gpo9JB6o3pSGr",
                 aws_secret_access_key="1Qug5H73R3kP8boIHvdVcFtcb1jU9GRWnlmMpx0g"
             )
@@ -340,9 +340,10 @@ def upload_to_s3(json_data: str):
             url = client.generate_presigned_url(
                 'get_object',
                 Params={'Bucket': "temp", 'Key': file_name},
-                ExpiresIn=3600
+                ExpiresIn=36000
             )
-            return url.replace("http://100.85.52.31:9000" or "", "https://s3-api.siat-mic.com")
+            # return url.replace("http://100.85.52.31:9000" or "", "https://s3-api.siat-mic.com")
+            return url
         
         except Exception as e:
             # print(e)
@@ -409,7 +410,6 @@ def get_latest_exp_log():
         # 找到最新修改的文件
         latest_file = pl_files[-1]
         res = f"找到最新的PL数据文件: {latest_file}"
-        # print(res)
         return res
 
     pl_latest = get_pl_latest_file()
@@ -417,6 +417,11 @@ def get_latest_exp_log():
 
     return pl_latest + "\n" + uv_latest
 
+
+def read_data():
+    get_latest_exp_log()
+
+
 def default_func():
     return "Approved. Proceed as planned!"
 
@@ -505,8 +510,8 @@ def sendScheme2RobotWorkstation(task_id: str, scheme_url: str):
         return {"status": "error", "message": f"Error reading scheme_url: {e}"}
     
     import requests
-    # url = "http://100.122.132.69:50000/sendScheme2RobotPlatform"
-    url = "http://localhost:50000/sendScheme2RobotWorkstation"
+    url = "http://100.122.132.69:50000/sendScheme2RobotPlatform"
+    # url = "http://localhost:50000/sendScheme2RobotWorkstation"
     data = {"status": "ok"}
     try:
         response = requests.post(url, json=robot_scheme)
@@ -579,8 +584,8 @@ def sendScheme2MobileRobot(task_id: str, scheme_url: str):
         return {"status": "error", "message": f"Error reading scheme_url: {e}"}
     
     import requests
-    # url = "http://100.122.132.69:50000/sendScheme2RobotPlatform"
-    url = "http://localhost:50000/sendScheme2MobileRobot"
+    url = "http://100.122.132.69:50000/sendScheme2RobotPlatform"
+    # url = "http://localhost:50000/sendScheme2MobileRobot"
     data = {"status": "ok"}
     try:
         response = requests.post(url, json=robot_scheme)
@@ -590,5 +595,42 @@ def sendScheme2MobileRobot(task_id: str, scheme_url: str):
         print(f"Error sending scheme to robot workstation: {e}")
         return None
 
+def readPLdata():
+    def list_to_markdown_table(data):
+        # 拆分表头和数据行
+        headers = data[0].split("\t")  # 假设列之间用制表符（\t）分隔
+        rows = [line.split("\t") for line in data[1:]]
+
+        # 创建表头部分
+        header_row = "| " + " | ".join(headers) + " |"
+        separator_row = "| " + " | ".join(["---"] * len(headers)) + " |"
+
+        # 创建数据行部分
+        data_rows = []
+        for row in rows:
+            data_rows.append("| " + " | ".join(row) + " |")
+
+        # 合并所有部分
+        markdown_table = "\n".join([header_row, separator_row] + data_rows)
+        return markdown_table
+
+    import requests
+    url = "http://100.122.132.69:50000/getPLdata"
+    try:
+        response = requests.get(url)
+        response.raise_for_status()
+        scheme_content = response.text
+        table = scheme_content.split("Data Points\r\n")[0].split("Peaks\r\n")[-1].split("\r\n")
+        table = [t for t in table if t.strip()]
+        md_table = list_to_markdown_table(table)
+        max_pl = table[1].split('\t')[2]
+        return md_table + "\n" + f"The max PL value in this expriment is {max_pl}"
+    
+    except Exception as e:
+        print(e)
+
+
 if __name__ == "__main__":
-    print(sendScheme2RobotWorkstation(task_id="task_20250122133404", scheme_url="https://s3-api.siat-mic.com/temp/robotExprimentScheme_task_20250122133404.json?AWSAccessKeyId=9bUtQL1Gpo9JB6o3pSGr&Signature=o9SGRdIN6h6%2BL8V7BCYALTc5s8k%3D&Expires=1737527678"))
+    # print(sendScheme2RobotWorkstation(task_id="20250225173342", scheme_url="http://100.85.52.31:9000/temp/robotExprimentScheme_task_20250225173342.json?AWSAccessKeyId=9bUtQL1Gpo9JB6o3pSGr&Signature=bkfes1Nv%2BVzwiUj05p7uHk%2B0P9g%3D&Expires=1740512067"))
+    # print(get_latest_exp_log())
+    print(readPLdata())