调整agent布局

_backend/constant.py

@@ -1,13 +1,14 @@
 from pathlib import Path
 import os
+from autogen_ext.code_executors.docker import DockerCommandLineCodeExecutor
 
 
 # Define your API keys and configurations
 OPENAI_API_KEY = "sk-4aJj5ygdQ9rw6lS6920712Ef9bB848439522E72318439eCd"
 OPENAI_BASE_URL = "http://8.218.238.241:17935/v1"
 
-MODEL = "chatgpt-4o-latest"
-# MODEL = "gpt-4o-2024-11-20"
+# MODEL = "chatgpt-4o-latest"
+MODEL = "gpt-4o-2024-11-20"
 # MODEL = "deepseek-chat"
 # config_list = [{"model": MODEL, "api_key": OPENAI_API_KEY, "base_url": OPENAI_BASE_URL, "temperature": 0}]
 config_list = [{"model": MODEL, "api_key": OPENAI_API_KEY, "base_url": OPENAI_BASE_URL}]
@@ -16,7 +17,7 @@ SILENT = True # 关闭嵌套智能体的输出
 STREAM = True # stream on console
 CACHE = None # None 就是关闭 41是默认值开启
 
-current_path = os.path.dirname(os.path.abspath(__file__))
-WORK_DIR = Path(current_path, ".coding")
-WORK_DIR.mkdir(exist_ok=True)
-
+WORK_DIR = os.path.join(os.path.dirname(os.path.abspath(__file__)), ".coding")
+if not os.path.exists(WORK_DIR):
+    os.mkdir(WORK_DIR)
+code_executor = DockerCommandLineCodeExecutor(bind_dir=Path(WORK_DIR))

_backend/custom.py

@@ -120,7 +120,6 @@ class SocietyOfMindAgent(BaseChatAgent):
     ) -> AsyncGenerator[AgentEvent | ChatMessage | Response, None]:
         # Prepare the task for the team of agents.
         task = list(messages)
-        task = [t for t in task if t.source!="PlanningAgent"][-2:]
 
         # Run the team of agents.
         result: TaskResult | None = None

_backend/engineer_team.py

@@ -1,13 +1,15 @@
-import asyncio
+import os
 from typing import Sequence
-from autogen_agentchat.agents import AssistantAgent, SocietyOfMindAgent
+from autogen_agentchat.agents import AssistantAgent, 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_agentchat.ui import Console
 from autogen_ext.models.openai import OpenAIChatCompletionClient
-from constant import MODEL, OPENAI_API_KEY, OPENAI_BASE_URL
-from tools import retrieval_from_knowledge_base, search_from_oqmd_by_composition, scheme_convert_to_json
+from constant import MODEL, OPENAI_API_KEY, OPENAI_BASE_URL, WORK_DIR
+from tools import retrieval_from_knowledge_base, search_from_oqmd_by_composition, scheme_convert_to_json, send_instruction_to_robot_platform, upload_to_s3
 # from custom import SocietyOfMindAgent
 
 model_client = OpenAIChatCompletionClient(
@@ -22,7 +24,7 @@ model_client = OpenAIChatCompletionClient(
     },
 )
 
-def create_engineer_team() -> SelectorGroupChat | RoundRobinGroupChat:
+def create_engineer_team() -> SelectorGroupChat | RoundRobinGroupChat | Swarm | SocietyOfMindAgent:
     planning_agent = AssistantAgent(
         "Engineer_PlanningAgent",
         description="An agent of Engineer team for planning tasks, this agent should be the first to engage when given a new task.",
@@ -30,42 +32,61 @@ def create_engineer_team() -> SelectorGroupChat | RoundRobinGroupChat:
         system_message="""
         You are a Engineer coordinator.
         Your job is coordinating material science research by delegating to specialized agents:
-            Software_Engineer: A Professional software engineers good at writing code to turn structured schemes in JSON format into Python code that can be executed by robots.
-            Structural_Engineer: A professional structural engineer who focus on converting natural language synthesis schemes to formatted synthesis schemes in JSON or XML format.
+            Structural_Engineer: A professional structural engineer who focus on converting natural language synthesis schemes to JSON or XML formated scheme, and then upload this JSON to S3 Storage.
+            Data_Engineer: A professional data engineer will use Python to implement various machine learning algorithms to analyze and visualize data.
+            SandBox_Env: A computer terminal that performs no other action than running Python scripts (provided to it quoted in ```python code blocks), or sh shell scripts (provided to it quoted in ```sh code blocks).
         Always send your plan first, then handoff to appropriate agent. Always handoff to a single agent at a time.
 
         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=["Software_Engineer", "Structural_Engineer"]
+        handoffs=["Software_Engineer", "Structural_Engineer", "Data_Engineer"]
     )
 
     structural_agent = AssistantAgent(
         "Structural_Engineer",
-        description="A professional structural engineer who focus on converting natural language synthesis schemes to formatted synthesis schemes in JSON or XML format.",
+        description="A professional structural engineer who focus on converting natural language synthesis schemes to JSON or XML formated scheme, and then upload this JSON to S3 Storage.",
         model_client=model_client,
         system_message="""
         你是一个Structural_Engineer.
-        你的任务是将下文/历史对话中的涉及到的合成方案转化为机器人可执行的标准JSON格式。
+        你的任务是先将下文/历史对话中的涉及到的合成方案转化为机器人可执行的标准JSON格式。
+        然后再将可执行的标准JSON文件上传到S3中方便机器人平台读取.
 
         Always handoff back to Engineer_PlanningAgent when JSON or XML is complete.
         """,
         handoffs=["Engineer_PlanningAgent"],
-        tools=[scheme_convert_to_json],
+        tools=[scheme_convert_to_json, upload_to_s3],
         reflect_on_tool_use=True
     )
 
-    software_agent = AssistantAgent(
-        "Software_Engineer",
-        description="A Professional software engineers good at writing code to turn structured schemes in JSON format into Python code that can be executed by robots.",
+    python_code_execution = PythonCodeExecutionTool(DockerCommandLineCodeExecutor(work_dir=WORK_DIR))
+    sandbox_env = AssistantAgent(
+        "sandbox_env",
+        description="A computer terminal that performs no other action than running Python scripts (provided to it quoted in ```python code blocks), or sh shell scripts (provided to it quoted in ```sh code blocks).",
         model_client=model_client,
         system_message="""
-        你是一个专业的Software_Engineer。
+        A computer terminal that performs no other action than running Python scripts (provided to it quoted in ```python code blocks), or sh shell scripts (provided to it quoted in ```sh code blocks).
 
         Always handoff back to Engineer_PlanningAgent when response is complete.
         """,
         handoffs=["Engineer_PlanningAgent"],
-        reflect_on_tool_use=True
+        reflect_on_tool_use=True,
+        tools=[python_code_execution]
+    )
+
+    software_agent = AssistantAgent(
+        "Software_Engineer",
+        description="A professional Python software engineer will use Python to implement tasks.",
+        model_client=model_client,
+        system_message="""
+        你是一个专业的Data_Engineer。
+        你的任务是使用Python代码完成用户的要求。
+    
+        Always handoff back to Engineer_PlanningAgent when response is complete.
+        """,
+        handoffs=["Engineer_PlanningAgent"],
+        reflect_on_tool_use=True,
+        #tools=[python_code_execution]
     )
 
     # The termination condition is a combination of text mention termination and max message termination.
@@ -76,7 +97,7 @@ def create_engineer_team() -> SelectorGroupChat | RoundRobinGroupChat:
     # termination = max_messages_termination
 
     team = Swarm(
-        participants=[planning_agent, structural_agent, software_agent],
+        participants=[planning_agent, structural_agent, software_agent, sandbox_env],
         termination_condition=termination
     )
 

_backend/main.py

@@ -7,9 +7,10 @@ from autogen_agentchat.teams import SelectorGroupChat, RoundRobinGroupChat
 from autogen_agentchat.ui import Console
 from autogen_agentchat.base import Handoff
 from autogen_ext.models.openai import OpenAIChatCompletionClient
-from constant import MODEL, OPENAI_API_KEY, OPENAI_BASE_URL
+from constant import MODEL, OPENAI_API_KEY, OPENAI_BASE_URL, code_executor
 from scientist_team import create_scientist_team
 from engineer_team import create_engineer_team
+from robot_platform import create_robot_team
 
 model_client = OpenAIChatCompletionClient(
     model=MODEL,
@@ -23,10 +24,11 @@ model_client = OpenAIChatCompletionClient(
     },
 )
 
-
 async def main(task: str = "") -> dict:
     scientist_team = create_scientist_team()
     engineer_team = create_engineer_team()
+    await code_executor.start()
+    robot_platform = create_robot_team(code_executor)
     
     result = {}
     user = UserProxyAgent("user", input_func=input)
@@ -43,24 +45,35 @@ async def main(task: str = "") -> dict:
             1. User: A human agent to whom you transfer information whenever you need to confirm your execution steps to a human.
             2. Engineer team: A team of professional engineers who are responsible for writing code, visualizing experimental schemes, converting experimental schemes to JSON, and more.
                 - The engineer team has the following members: 
-                2.1 Software engineer: Good at writing code to turn structured schemes in JSON format into Python code that can be executed by robots.
-                2.2 Structural engineer: Focus on converting natural language synthesis schemes to formatted synthesis schemes in JSON or XML format.
+                2.1 Structural engineer: A professional structural engineer who focus on converting natural language synthesis schemes to JSON or XML formated scheme, and then upload this JSON to S3 Storage.
+                2.2 Software_Engineer: A professional Python software engineer will use Python to implement tasks.
+                2.3 SandBox_Env: A computer terminal that performs no other action than running Python scripts (provided to it quoted in ```python code blocks), or sh shell scripts (provided to it quoted in ```sh code blocks).
             3. Scientist team: A professional team of material scientists who are mainly responsible for consulting on material synthesis, structure, application and properties.
                 - The scientist team has the following members: 
                 3.1 Synthesis Scientist: who is good at giving perfect and correct synthesis solutions.
                 3.2 Structure Scientist: focusing on agents of structural topics in materials science.
                 3.3 Property Scientist: focuses on physical and chemistry property topics in materials science.
                 3.4 Application Scientist: Focus on practical applications of materials, such as devices, chips, etc.
+            4. Robot Platform: A robotic platform is responsible for performing automated synthesis experiments, automated characterization experiments, and collecting experimental logs.
+                - The Robot Platform has the following members: 
+                4.1 RobotIO_Agent: The agent responsible for the input and output of the robot platform. Pass a structured JSON schema to the robot; Get the experiment log of the robot.
 
         You only plan and delegate tasks - you do not execute them yourself.
         
-        第一次回答时你需要初始化任务分配并按顺序执行，在后续的回答中重申你的任务分配，使用如下格式并利用Mermaid绘制流程图：
-        | Team_name   | Member_name   | sub-task                     |
-        | ----------- | ------------- | ---------------------------- |
-        | <team_name> | <member_name> | <brief sub-task description> |
+        回答时你需要初始化/更新如下任务分配表和Mermaid流程图，并按顺序执行，使用如下格式并利用：
+        | Team_name   | Member_name   | sub-task                             |
+        | ----------- | ------------- | ------------------------------------ |
+        | <team_name> | <member_name> | <status: brief sub-task description> |
         
         ```mermaid
-        graph
+        graph TD
+        User[User]
+        subgraph <team_name>
+            A1[<member_name>]
+        end
+        style xxx # 推荐多样的风格
+        ...
+        User --> A1
         ...
         ```
 
@@ -70,7 +83,7 @@ async def main(task: str = "") -> dict:
         **Next sub-task:**
         n. <team> : <subtask>
         
-        You can end with "USER" if you need to, which means you need human approval or other advice or instructions;
+        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".
         """,
@@ -87,17 +100,18 @@ async def main(task: str = "") -> dict:
     def selector_func(messages: Sequence[AgentEvent | ChatMessage]) -> str | None:
         if messages[-1].source != planning_agent.name:
             return planning_agent.name # Always return to the planning agent after the other agents have spoken.
-        elif "USER" in messages[-1].content:
+        elif "HUMAN" in messages[-1].content:
             return user.name
         return None
 
     team = SelectorGroupChat(
-        [planning_agent, user, scientist_team, engineer_team],
+        [planning_agent, user, scientist_team, engineer_team, robot_platform],
         model_client=model_client, # Use a smaller model for the selector.
         termination_condition=termination,
         selector_func=selector_func,
     )
     await Console(team.run_stream(task=task))
+    await code_executor.stop()
     # async for message in team.run_stream(task=task):
     #     if isinstance(message, TextMessage):
     #         print(f"----------------{message.source}----------------\n {message.content}")

_backend/robot_platform.py

@@ -0,0 +1,82 @@
+import asyncio
+from typing import Sequence
+from autogen_agentchat.agents import AssistantAgent, 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_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 send_instruction_to_robot_platform, get_latest_exp_log
+from custom import SocietyOfMindAgent
+
+from pathlib import Path
+
+model_client = OpenAIChatCompletionClient(
+    model=MODEL,
+    base_url=OPENAI_BASE_URL,
+    api_key=OPENAI_API_KEY,
+    model_info={
+        "vision": True,
+        "function_calling": True,
+        "json_output": True,
+        "family": "unknown",
+    },
+)
+
+def create_robot_team(code_executor) -> SelectorGroupChat | RoundRobinGroupChat | Swarm:
+    planning_agent = AssistantAgent(
+        "Robot_PlanningAgent",
+        description="An agent of Robot team for planning tasks, this agent should be the first to engage when given a new task.",
+        model_client=model_client,
+        system_message="""
+        You are a robot manager.
+        Your job is coordinating material science research by delegating to specialized agents:
+            RobotPlatform_Agent: The agent controls the robot platform to automate the experiment by calling the function xxx to send the experiment flow to the robot and execute it.
+            MobileRobot_Agent: This agent controls the mobile robot by calling the function xxx to assist the robot platform in completing the experiment.
+            DataCollector_Agent: This Agent will collect the data after the robot automation experiment, mainly including PL, UV and so on.
+        Always send your plan first, then handoff to appropriate agent. Always handoff to a single agent at a time.
+
+        After all tasks are completed, the member scientist 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=["RobotPlatform_Agent", "MobileRobot_Agent", "DataCollector_Agent"]
+    )
+
+    robot_agent = AssistantAgent(
+        "RobotPlatform_Agent",
+        description="The agent controls the robot platform to automate the experiment by calling the function xxx to send the experiment flow to the robot and execute it.",
+        model_client=model_client,
+        system_message="""
+        你是一个RobotIO_Agent。
+        The agent controls the robot platform to automate the experiment by calling the function xxx to send the experiment flow to the robot and execute it.
+    
+        Always handoff back to Robot_PlanningAgent when response is complete.
+        """,
+        handoffs=["Robot_PlanningAgent"],
+        reflect_on_tool_use=True,
+        tools=[send_instruction_to_robot_platform, get_latest_exp_log] 
+    )
+
+
+    # The termination condition is a combination of text mention termination and max message termination.
+    handoff_termination = HandoffTermination("Robot_PlanningAgent")
+    text_mention_termination = TextMentionTermination("APPROVE")
+    max_messages_termination = MaxMessageTermination(max_messages=50)
+    termination = text_mention_termination | max_messages_termination | handoff_termination
+    # termination = max_messages_termination
+
+    team = Swarm(
+        participants=[planning_agent, robot_agent],
+        termination_condition=termination
+    )
+
+    robot_platform = SocietyOfMindAgent(
+        name="robot_platform", 
+        team=team, 
+        description="A robotic platform is responsible for performing automated synthesis experiments, automated characterization experiments, and collecting experimental logs.",
+        model_client=model_client)
+    return robot_platform

_backend/robot_server.py

@@ -0,0 +1,45 @@
+from fastapi import FastAPI, HTTPException
+from pydantic import BaseModel
+import uvicorn
+
+app = FastAPI()
+
+class UrlRequest(BaseModel):
+    url: str
+
+@app.post("/receive")
+async def receive_url(url_request: UrlRequest):
+    """
+    接收并处理URL
+    
+    Args:
+        url_request (UrlRequest): 包含URL的请求体
+        
+    Returns:
+        dict: 包含处理结果的响应
+    """
+    try:
+        # 这里可以添加处理URL的逻辑
+        print(f"Received URL: {url_request.url}")
+        
+        # 返回成功响应
+        return {
+            "status": "success",
+            "message": "URL received successfully",
+            "url": url_request.url
+        }
+    except Exception as e:
+        raise HTTPException(status_code=500, detail=str(e))
+
+def start_server(host: str = "0.0.0.0", port: int = 8030):
+    """
+    启动FastAPI服务器
+    
+    Args:
+        host (str): 服务器主机地址
+        port (int): 服务器端口号
+    """
+    uvicorn.run(app, host=host, port=port)
+
+if __name__ == "__main__":
+    start_server()

_backend/scientist_team.py

@@ -22,7 +22,7 @@ model_client = OpenAIChatCompletionClient(
     },
 )
 
-def create_scientist_team() -> SelectorGroupChat | RoundRobinGroupChat | Swarm | AssistantAgent:
+def create_scientist_team() -> SelectorGroupChat | RoundRobinGroupChat | Swarm | SocietyOfMindAgent:
     planning_agent = AssistantAgent(
         "Scientist_PlanningAgent",
         description="An agent of Scientist team for planning tasks, this agent should be the first to engage when given a new task.",

_backend/tools.py

@@ -128,5 +128,190 @@ def scheme_convert_to_json():
         12. step_output 类型: 字符串; 说明: 步骤的输出标识符，用于后续步骤的输入。限制: 标识符应唯一且有意义。
 """
 
+def send_instruction_to_robot_platform():
+    """从S3获取最新的json文件并返回其URL链接
+    
+    Returns:
+        str: 最新json文件的预签名URL
+    """
+    import boto3
+    from botocore.exceptions import NoCredentialsError
+    
+    # S3配置
+    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"
+    bucket_name = "temp"
+
+    try:
+        # 创建S3客户端
+        s3 = boto3.client(
+            's3',
+            endpoint_url=endpoint_url,
+            aws_access_key_id=aws_access_key_id,
+            aws_secret_access_key=aws_secret_access_key
+        )
+
+        # 列出bucket中的所有json文件
+        response = s3.list_objects_v2(
+            Bucket=bucket_name,
+            Prefix='',
+            Delimiter='/'
+        )
+
+        # 过滤出json文件并按最后修改时间排序
+        json_files = [
+            obj for obj in response.get('Contents', [])
+            if obj['Key'].endswith('.json')
+        ]
+        json_files.sort(key=lambda x: x['LastModified'], reverse=True)
+
+        if not json_files:
+            return "No JSON files found in S3 bucket"
+
+        # 获取最新文件
+        latest_file = json_files[0]
+        
+        # 生成预签名URL
+        url = s3.generate_presigned_url(
+            'get_object',
+            Params={
+                'Bucket': bucket_name,
+                'Key': latest_file['Key']
+            },
+            ExpiresIn=3600  # URL有效期1小时
+        )
+        
+        # 将内部URL转换为外部可访问URL
+        external_url = url.replace("http://100.85.52.31:9000", "https://s3-api.siat-mic.com")
+        
+        # 发送URL到FastAPI服务器
+        try:
+            response = requests.post(
+                "http://localhost:8030/receive",
+                json={"url": external_url}
+            )
+            response.raise_for_status()
+            return external_url
+        except requests.exceptions.RequestException as e:
+            return f"Error sending URL to server: {str(e)}"
+
+    except NoCredentialsError:
+        return "Credentials not available"
+    except Exception as e:
+        return f"Error: {str(e)}"
+
+
+def upload_to_s3(json_data: str):
+    import json
+    import re
+    import subprocess
+    import sys
+    import tempfile
+    import datetime
+
+    def install_boto3():
+        try:
+            # 检查 boto3 是否已安装
+            import boto3
+            print("boto3 已安装。")
+        except ImportError:
+            # 如果未安装，动态安装 boto3
+            print("正在安装 boto3...")
+            subprocess.check_call([sys.executable, "-m", "pip", "install", "boto3"])
+            print("boto3 安装完成。")
+
+    def handle_minio_upload(file_path: str, file_name: str) -> str:
+        """统一处理MinIO上传"""
+        import boto3
+        try:
+            client = boto3.client(
+                's3',
+                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"
+            )
+            client.upload_file(file_path, "temp", file_name, ExtraArgs={"ACL": "private"})
+            
+            # 生成预签名 URL
+            url = client.generate_presigned_url(
+                'get_object',
+                Params={'Bucket': "temp", 'Key': file_name},
+                ExpiresIn=3600
+            )
+            return url.replace("http://100.85.52.31:9000" or "", "https://s3-api.siat-mic.com")
+        
+        except Exception as e:
+            print(e)
+            return e
+        
+    install_boto3()
+    # 去掉可能存在的 ```json 和 ``` 标记
+    json_data_cleaned = re.sub(r'```json|```', '', json_data).strip()
+    try:
+        # 尝试解析清理后的JSON数据
+        data = json.loads(json_data_cleaned)
+        # print("解析后的JSON数据:", data)
+        with tempfile.NamedTemporaryFile(mode='w', delete=False) as temp_file:
+            try:
+                json.dump(data, temp_file, indent=4, ensure_ascii=False)
+                temp_file.flush()  # 确保数据写入文件
+                timestamp = datetime.datetime.now().strftime("%Y%m%d%H%M%S")
+                file_name = f"robot_expriment_results_{timestamp}.json"
+                url = handle_minio_upload(temp_file.name, file_name)
+                print(f"文件上传成功, 唯一链接为{url}, 请将该URL记下来并传递给机器人平台。")
+            except Exception as e:
+                print(f"写入临时文件或上传文件时出错: {e}")
+                raise  # 重新抛出异常以便上层调用者处理
+    except json.JSONDecodeError as e:
+        print(f"JSON解析错误: {e}")
+        raise  # 重新抛出异常以便上层调用者处理
+
+def get_latest_exp_log():
+    def get_uv_latest_file():
+        import os
+        import glob
+        # UV数据缓存文件夹路径 (请将此路径修改为实际的文件夹路径)
+        current_folder = os.path.dirname(os.path.abspath(__file__))
+        folder_path = os.path.join(current_folder, 'data/UV/')
+
+        # 查找文件夹中的所有 .wls 文件
+        uv_files = sorted(glob.glob(os.path.join(folder_path, '*.[Tt][Xx][Tt]')))
+
+        if not uv_files:
+            res = f"ERROR: 缓存文件夹{current_folder}中没有找到任何UV文件"
+            return res
+
+        # 找到最新修改的文件
+        latest_file = uv_files[-1]
+        res = f"找到最新的UV数据文件: {latest_file}"
+        
+        return res
+    
+    def get_pl_latest_file():
+        import os
+        import glob
+
+        current_folder = os.path.dirname(os.path.abspath(__file__))
+        folder_path = os.path.join(current_folder, 'data/PL/')
+
+        # 查找文件夹中的所有 .txt 或 .TXT 文件
+        pl_files = sorted(glob.glob(os.path.join(folder_path, '*.[Tt][Xx][Tt]')))
+
+        if not pl_files:
+            res = f"ERROR: 缓存文件夹{current_folder}中没有找到任何PL文件"
+            return res
+
+        # 找到最新修改的文件
+        latest_file = pl_files[-1]
+        res = f"找到最新的PL数据文件: {latest_file}"
+        # print(res)
+        return res
+
+    pl_latest = get_pl_latest_file() 
+    uv_latest = get_uv_latest_file()
+
+    return pl_latest + "\n" + uv_latest
+
 def default_func():
-    return "Approved. Proceed as planned!"
+    return "Approved. Proceed as planned!"

_backend/utils.py

@@ -0,0 +1,72 @@
+"""
+Author: Yutang LI
+Institution: SIAT-MIC
+Contact: yt.li2@siat.ac.cn
+"""
+
+import os
+import boto3
+import logging
+from typing import Optional, Dict
+from pydantic import Field
+from pydantic_settings import BaseSettings
+
+logger = logging.getLogger(__name__)
+
+class Settings(BaseSettings):
+    # Model
+    openai_api_key: Optional[str] = Field(None, env="OPENAI_API_KEY")
+    openai_base_url: Optional[str] = Field(None, env="OPENAI_BASE_URL")
+    model_name: Optional[str] = Field(None, env="MODEL")
+
+    # Path
+    coding_path: Optional[str] = Field(None, env=os.path.join(os.path.dirname(os.path.abspath(__file__)), "CODING_DIR"))
+    
+    # MinIO
+    minio_endpoint: Optional[str] = Field(None, env="MINIO_ENDPOINT")
+    internal_minio_endpoint: Optional[str] = Field(None, env="INTERNAL_MINIO_ENDPOINT")
+    minio_access_key: Optional[str] = Field(None, env="MINIO_ACCESS_KEY")
+    minio_secret_key: Optional[str] = Field(None, env="MINIO_SECRET_KEY")
+    minio_bucket: Optional[str] = Field("temp", env="MINIO_BUCKET")
+    
+    class Config:
+        env_file = ".env"
+        env_file_encoding = "utf-8"
+
+
+# 初始化配置
+settings = Settings()
+
+def handle_minio_error(e: Exception) -> Dict[str, str]:
+    """处理MinIO相关错误"""
+    logger.error(f"MinIO operation failed: {str(e)}")
+    return {
+        "status": "error",
+        "data": f"MinIO operation failed: {str(e)}"
+    }
+
+def get_minio_client(settings: Settings):
+    """获取MinIO客户端"""
+    return boto3.client(
+        's3',
+        endpoint_url=settings.internal_minio_endpoint or settings.minio_endpoint,
+        aws_access_key_id=settings.minio_access_key,
+        aws_secret_access_key=settings.minio_secret_key
+    )
+
+def handle_minio_upload(file_path: str, file_name: str) -> str:
+    """统一处理MinIO上传"""
+    try:
+        client = get_minio_client(settings)
+        client.upload_file(file_path, settings.minio_bucket, file_name, ExtraArgs={"ACL": "private"})
+        
+        # 生成预签名 URL
+        url = client.generate_presigned_url(
+            'get_object',
+            Params={'Bucket': settings.minio_bucket, 'Key': file_name},
+            ExpiresIn=3600
+        )
+        return url.replace(settings.internal_minio_endpoint or "", settings.minio_endpoint)
+    
+    except Exception as e:
+        return handle_minio_error(e)