Upload 4 files

AWorld-main/aworlddistributed/aworldspace/utils/loader.py

@@ -0,0 +1,197 @@
+import importlib.util
+import json
+import logging
+import os
+import subprocess
+import sys
+import traceback
+from aworldspace.base import AGENT_SPACE
+import aworld.trace as trace  # noqa
+
+from config import AGENTS_DIR
+
+if not os.path.exists(AGENTS_DIR):
+    os.makedirs(AGENTS_DIR)
+
+PIPELINES = {}
+PIPELINE_MODULES = {}
+
+def get_all_pipelines():
+    pipelines = {}
+    for pipeline_id in PIPELINE_MODULES.keys():
+        pipeline = PIPELINE_MODULES[pipeline_id]
+
+        if hasattr(pipeline, "type"):
+            if pipeline.type == "manifold":
+                manifold_pipelines = []
+
+                # Check if pipelines is a function or a list
+                if callable(pipeline.pipelines):
+                    manifold_pipelines = pipeline.pipelines()
+                else:
+                    manifold_pipelines = pipeline.pipelines
+
+                for p in manifold_pipelines:
+                    manifold_pipeline_id = f'{pipeline_id}.{p["id"]}'
+
+                    manifold_pipeline_name = p["name"]
+                    if hasattr(pipeline, "name"):
+                        manifold_pipeline_name = (
+                            f"{pipeline.name}{manifold_pipeline_name}"
+                        )
+
+                    pipelines[manifold_pipeline_id] = {
+                        "module": pipeline_id,
+                        "type": pipeline.type if hasattr(pipeline, "type") else "pipe",
+                        "id": manifold_pipeline_id,
+                        "name": manifold_pipeline_name,
+                        "valves": (
+                            pipeline.valves if hasattr(pipeline, "valves") else None
+                        ),
+                    }
+            if pipeline.type == "filter":
+                pipelines[pipeline_id] = {
+                    "module": pipeline_id,
+                    "type": (pipeline.type if hasattr(pipeline, "type") else "pipe"),
+                    "id": pipeline_id,
+                    "name": (
+                        pipeline.name if hasattr(pipeline, "name") else pipeline_id
+                    ),
+                    "pipelines": (
+                        pipeline.valves.pipelines
+                        if hasattr(pipeline, "valves")
+                        and hasattr(pipeline.valves, "pipelines")
+                        else []
+                    ),
+                    "priority": (
+                        pipeline.valves.priority
+                        if hasattr(pipeline, "valves")
+                        and hasattr(pipeline.valves, "priority")
+                        else 0
+                    ),
+                    "valves": pipeline.valves if hasattr(pipeline, "valves") else None,
+                }
+        else:
+            pipelines[pipeline_id] = {
+                "module": pipeline_id,
+                "type": (pipeline.type if hasattr(pipeline, "type") else "pipe"),
+                "id": pipeline_id,
+                "name": (pipeline.name if hasattr(pipeline, "name") else pipeline_id),
+                "valves": pipeline.valves if hasattr(pipeline, "valves") else None,
+            }
+
+    return pipelines
+
+
+def parse_frontmatter(content):
+    frontmatter = {}
+    for line in content.split("\n"):
+        if ":" in line:
+            key, value = line.split(":", 1)
+            frontmatter[key.strip().lower()] = value.strip()
+    return frontmatter
+
+
+def install_frontmatter_requirements(requirements):
+    if requirements:
+        req_list = [req.strip() for req in requirements.split(",")]
+        for req in req_list:
+            print(f"Installing requirement: {req}")
+            subprocess.check_call([sys.executable, "-m", "pip", "install", req])
+    else:
+        print("No requirements found in frontmatter.")
+
+
+async def load_module_from_path(module_name, module_path):
+
+    try:
+        # Read the module content
+        with open(module_path, "r") as file:
+            content = file.read()
+
+        # Parse frontmatter
+        frontmatter = {}
+        if content.startswith('"""'):
+            end = content.find('"""', 3)
+            if end != -1:
+                frontmatter_content = content[3:end]
+                frontmatter = parse_frontmatter(frontmatter_content)
+
+        # Install requirements if specified
+        if "requirements" in frontmatter:
+            install_frontmatter_requirements(frontmatter["requirements"])
+
+        # Load the module
+        spec = importlib.util.spec_from_file_location(module_name, module_path)
+        module = importlib.util.module_from_spec(spec)
+        spec.loader.exec_module(module)
+        logging.info(f"Loaded module start: {module.__name__}")
+        if hasattr(module, "Pipeline"):
+            return module.Pipeline()
+        else:
+            logging.info(f"Loaded module failed: {module.__name__ } No Pipeline class found")
+            raise Exception("No Pipeline class found")
+    except Exception as e:
+        logging.info(f"Error loading module: {module_name}, error is {e}")
+        traceback.print_exc()
+        # Move the file to the error folder
+        failed_pipelines_folder = os.path.join(AGENTS_DIR, "failed")
+        if not os.path.exists(failed_pipelines_folder):
+            os.makedirs(failed_pipelines_folder)
+
+        # failed_file_path = os.path.join(failed_pipelines_folder, f"{module_name}.py")
+        # if module_path.__contains__(PIPELINES_DIR):
+        #     os.rename(module_path, failed_file_path)
+        print(e)
+    return None
+
+
+async def load_modules_from_directory(directory):
+    logging.info(f"load_modules_from_directory: {directory}")
+    global PIPELINE_MODULES
+
+    for filename in os.listdir(directory):
+        if filename.endswith(".py"):
+            module_name = filename[:-3]  # Remove the .py extension
+            module_path = os.path.join(directory, filename)
+
+            # Create subfolder matching the filename without the .py extension
+            subfolder_path = os.path.join(directory, module_name)
+            if not os.path.exists(subfolder_path):
+                os.makedirs(subfolder_path)
+                logging.info(f"Created subfolder: {subfolder_path}")
+
+            # Create a valves.json file if it doesn't exist
+            valves_json_path = os.path.join(subfolder_path, "valves.json")
+            if not os.path.exists(valves_json_path):
+                with open(valves_json_path, "w") as f:
+                    json.dump({}, f)
+                logging.info(f"Created valves.json in: {subfolder_path}")
+
+            pipeline = await load_module_from_path(module_name, module_path)
+            if pipeline:
+                # Overwrite pipeline.valves with values from valves.json
+                if os.path.exists(valves_json_path):
+                    with open(valves_json_path, "r") as f:
+                        valves_json = json.load(f)
+                        if hasattr(pipeline, "valves"):
+                            ValvesModel = pipeline.valves.__class__
+                            # Create a ValvesModel instance using default values and overwrite with valves_json
+                            combined_valves = {
+                                **pipeline.valves.model_dump(),
+                                **valves_json,
+                            }
+                            valves = ValvesModel(**combined_valves)
+                            pipeline.valves = valves
+
+                            logging.info(f"Updated valves for module: {module_name}")
+
+                pipeline_id = pipeline.id if hasattr(pipeline, "id") else module_name
+                PIPELINE_MODULES[pipeline_id] = pipeline
+
+                logging.info(f"Loaded module success: {module_name}")
+            else:
+                logging.warning(f"No Pipeline class found in {module_name}")
+
+    AGENT_SPACE.agent_modules = PIPELINE_MODULES
+    AGENT_SPACE.agents_meta = get_all_pipelines()

AWorld-main/aworlddistributed/aworldspace/utils/log.py

@@ -0,0 +1,75 @@
+import logging
+import os
+from datetime import datetime
+
+from aworld.models.model_response import ModelResponse
+
+from base import AworldTask, AworldTaskResult
+from config import ROOT_LOG
+
+
+class TaskLogger:
+    """任务提交日志记录器"""
+
+    def __init__(self, log_file: str = "aworld_task_submissions.log"):
+        self.log_file = os.path.join(ROOT_LOG, 'task_logs' , log_file)
+        self._ensure_log_file_exists()
+
+    def _ensure_log_file_exists(self):
+        """确保日志文件存在"""
+        if not os.path.exists(self.log_file):
+            os.makedirs(os.path.dirname(self.log_file), exist_ok=True)
+            with open(self.log_file, 'w', encoding='utf-8') as f:
+                f.write("# Aworld Task Submission Log\n")
+                f.write(
+                    "# Format: [timestamp] task_id | agent_id | server | status | agent_answer | correct_answer | is_correct | details\n\n")
+
+    def log_task_submission(self, task: AworldTask, status: str, details: str = "",
+                            task_result: AworldTaskResult = None):
+        """记录任务提交日志"""
+        timestamp = datetime.now().strftime("%Y-%m-%d %H:%M:%S")
+        log_entry = f"[{timestamp}] {task.task_id} | {task.agent_id} | {task.node_id} | {status} | {task_result.data.get('agent_answer') if task_result and task_result.data else None} | {task_result.data.get('correct_answer') if task_result and task_result.data else None} | {task_result.data.get('gaia_correct') if task_result and task_result.data else None} |{details}\n"
+
+        try:
+            with open(self.log_file, 'a', encoding='utf-8') as f:
+                f.write(log_entry)
+        except Exception as e:
+            logging.error(f"Failed to write task submission log: {e}")
+
+    def log_task_result(self, task: AworldTask, result: ModelResponse):
+        try:
+            date_str = datetime.now().strftime("%Y%m%d")
+            result_dir = os.path.join(ROOT_LOG, 'task_logs', 'result', date_str)
+            os.makedirs(result_dir, exist_ok=True)
+
+            md_file = f"{result_dir}/{task.task_id}.md"
+
+            content_parts = []
+            if hasattr(result, 'content') and result.content:
+                if isinstance(result.content, list):
+                    content_parts.extend(result.content)
+                else:
+                    content_parts.append(str(result.content))
+
+            file_exists = os.path.exists(md_file)
+            with open(md_file, 'a', encoding='utf-8') as f:
+                if not file_exists:
+                    f.write(f"# Task Result: {task.task_id}\n\n")
+                    f.write(f"**Agent ID:** {task.agent_id}\n\n")
+                    f.write(f"**Timestamp:** {datetime.now().strftime('%Y-%m-%d %H:%M:%S')}\n\n")
+                    f.write("## Content\n\n")
+
+                if content_parts:
+                    for i, content in enumerate(content_parts, 1):
+                        f.write(f"{content}\n\n")
+                else:
+                    f.write("No content available.\n\n")
+
+            return md_file
+
+        except Exception as e:
+            logging.error(f"Failed to write task result log: {e}")
+            return None
+
+
+task_logger = TaskLogger(log_file=f"aworld_task_submissions_{datetime.now().strftime('%Y%m%d')}.log")

AWorld-main/aworlddistributed/aworldspace/utils/mcp_utils.py

@@ -0,0 +1,199 @@
+import os
+
+
+def load_all_mcp_config():
+    return {
+        "mcpServers": {
+            "e2b-server": {
+                "command": "npx",
+                "args": [
+                    "-y",
+                    "@e2b/mcp-server"
+                ],
+                "env": {
+                    "E2B_API_KEY": os.environ["E2B_API_KEY"],
+                    "SESSION_REQUEST_CONNECT_TIMEOUT": "60"
+                }
+            },
+            "filesystem": {
+                "command": "npx",
+                "args": [
+                    "-y",
+                    "@modelcontextprotocol/server-filesystem",
+                    "${FILESYSTEM_SERVER_WORKDIR}"
+                ]
+            },
+            "terminal-controller": {
+                "command": "python",
+                "args": [
+                    "-m",
+                    "terminal_controller"
+                ],
+                "env": {
+                    "SESSION_REQUEST_CONNECT_TIMEOUT": "300"
+                }
+            },
+            "calculator": {
+                "command": "python",
+                "args": [
+                    "-m",
+                    "mcp_server_calculator"
+                ],
+                "env": {
+                    "SESSION_REQUEST_CONNECT_TIMEOUT": "20"
+                }
+            },
+            "excel": {
+                "command": "uvx",
+                "args": ["excel-mcp-server", "stdio"],
+                "env": {
+                    "EXCEL_MCP_PAGING_CELLS_LIMIT": "4000",
+                    "SESSION_REQUEST_CONNECT_TIMEOUT": "120"
+                }
+            },
+            "google-search": {
+                "command": "npx",
+                "args": [
+                    "-y",
+                    "@adenot/mcp-google-search"
+                ],
+                "env": {
+                    "GOOGLE_API_KEY": os.environ["GOOGLE_API_KEY"],
+                    "GOOGLE_SEARCH_ENGINE_ID": os.environ["GOOGLE_CSE_ID"],
+                    "SESSION_REQUEST_CONNECT_TIMEOUT": "60"
+                }
+            },
+            "ms-playwright": {
+                "command": "npx",
+                "args": [
+                    "@playwright/mcp@latest",
+                    "--no-sandbox",
+                    "--headless",
+                    "--isolated"
+                ],
+                "env": {
+                    "PLAYWRIGHT_TIMEOUT": "120000",
+                    "SESSION_REQUEST_CONNECT_TIMEOUT": "120"
+                }
+            },
+            "audio_server": {
+                "command": "python",
+                "args": [
+                    "-m",
+                    "mcp_servers.audio_server"
+                ],
+                "env": {
+                    "AUDIO_LLM_API_KEY": os.environ["AUDIO_LLM_API_KEY"],
+                    "AUDIO_LLM_BASE_URL": os.environ["AUDIO_LLM_BASE_URL"],
+                    "AUDIO_LLM_MODEL_NAME": os.environ["AUDIO_LLM_MODEL_NAME"],
+                    "SESSION_REQUEST_CONNECT_TIMEOUT": "60"
+                }
+            },
+            "image_server": {
+                "command": "python",
+                "args": [
+                    "-m",
+                    "mcp_servers.image_server"
+                ],
+                "env": {
+                    "LLM_API_KEY": os.environ.get("LLM_API_KEY"),
+                    "LLM_MODEL_NAME": os.environ.get("LLM_MODEL_NAME"),
+                    "LLM_BASE_URL": os.environ.get("LLM_BASE_URL"),
+                    "SESSION_REQUEST_CONNECT_TIMEOUT": "60"
+                }
+            },
+            "youtube_server": {
+                "command": "python",
+                "args": [
+                    "-m",
+                    "mcp_servers.youtube_server"
+                ],
+                "env": {
+                    "CHROME_DRIVER_PATH": os.environ['CHROME_DRIVER_PATH'],
+                    "SESSION_REQUEST_CONNECT_TIMEOUT": "120"
+                }
+            },
+            "video_server": {
+                "command": "python",
+                "args": [
+                    "-m",
+                    "mcp_servers.video_server"
+                ],
+                "env": {
+                    "LLM_API_KEY": os.environ.get("LLM_API_KEY"),
+                    "LLM_MODEL_NAME": os.environ.get("LLM_MODEL_NAME"),
+                    "LLM_BASE_URL": os.environ.get("LLM_BASE_URL"),
+                    "SESSION_REQUEST_CONNECT_TIMEOUT": "60"
+                }
+            },
+            "search_server": {
+                "command": "python",
+                "args": [
+                    "-m",
+                    "mcp_servers.search_server"
+                ],
+                "env": {
+                    "GOOGLE_API_KEY": os.environ["GOOGLE_API_KEY"],
+                    "GOOGLE_CSE_ID": os.environ["GOOGLE_CSE_ID"],
+                    "SESSION_REQUEST_CONNECT_TIMEOUT": "60"
+                }
+            },
+            "download_server": {
+                "command": "python",
+                "args": [
+                    "-m",
+                    "mcp_servers.download_server"
+                ],
+                "env": {
+                    "SESSION_REQUEST_CONNECT_TIMEOUT": "120"
+                }
+            },
+            "document_server": {
+                "command": "python",
+                "args": [
+                    "-m",
+                    "mcp_servers.document_server"
+                ],
+                "env": {
+                    "SESSION_REQUEST_CONNECT_TIMEOUT": "120"
+                }
+            },
+            "browser_server": {
+                "command": "python",
+                "args": [
+                    "-m",
+                    "mcp_servers.browser_server"
+                ],
+                "env": {
+                    "LLM_API_KEY": os.environ.get("LLM_API_KEY"),
+                    "LLM_MODEL_NAME": os.environ.get("LLM_MODEL_NAME"),
+                    "LLM_BASE_URL": os.environ.get("LLM_BASE_URL"),
+                    "SESSION_REQUEST_CONNECT_TIMEOUT": "120"
+                }
+            },
+            "reasoning_server": {
+                "command": "python",
+                "args": [
+                    "-m",
+                    "mcp_servers.reasoning_server"
+                ],
+                "env": {
+                    "LLM_API_KEY": os.environ.get("LLM_API_KEY"),
+                    "LLM_MODEL_NAME": os.environ.get("LLM_MODEL_NAME"),
+                    "LLM_BASE_URL": os.environ.get("LLM_BASE_URL"),
+                    "SESSION_REQUEST_CONNECT_TIMEOUT": "120"
+                }
+            },
+            "e2b-code-server": {
+                "command": "python",
+                "args": [
+                    "-m",
+                    "mcp_servers.e2b_code_server"
+                ],
+                "env": {
+                    "E2B_API_KEY": os.environ["E2B_API_KEY"],
+                    "SESSION_REQUEST_CONNECT_TIMEOUT": "120"
+                }
+            },
+        }
+    }

AWorld-main/aworlddistributed/aworldspace/utils/utils.py

@@ -0,0 +1,344 @@
+import json
+import re
+import string
+from pathlib import Path
+from typing import Any, Dict, List, Optional
+
+from loguru import logger
+from tabulate import tabulate
+
+
+def normalize_str(input_str, remove_punct=True) -> str:
+    no_spaces = re.sub(r"\s", "", input_str)
+    if remove_punct:
+        translator = str.maketrans("", "", string.punctuation)
+        return no_spaces.lower().translate(translator)
+    else:
+        return no_spaces.lower()
+
+
+def split_string(s: str, char_list: Optional[List[str]] = None) -> list[str]:
+    if char_list is None:
+        char_list = [",", ";"]
+    pattern = f"[{''.join(char_list)}]"
+    return re.split(pattern, s)
+
+
+def normalize_number_str(number_str: str) -> float:
+    for char in ["$", "%", ","]:
+        number_str = number_str.replace(char, "")
+    try:
+        return float(number_str)
+    except ValueError:
+        logger.error(f"String {number_str} cannot be normalized to number str.")
+        return float("inf")
+
+
+def question_scorer(model_answer: str, ground_truth: str) -> bool:
+    def is_float(element: Any) -> bool:
+        try:
+            float(element)
+            return True
+        except ValueError:
+            return False
+
+    try:
+        if is_float(ground_truth):
+            logger.info(f"Evaluating {model_answer} as a number.")
+            normalized_answer = normalize_number_str(model_answer)
+            return normalized_answer == float(ground_truth)
+
+        elif any(char in ground_truth for char in [",", ";"]):
+            logger.info(f"Evaluating {model_answer} as a comma separated list.")
+            gt_elems = split_string(ground_truth)
+            ma_elems = split_string(model_answer)
+
+            if len(gt_elems) != len(ma_elems):
+                logger.warning("Answer lists have different lengths, returning False.")
+                return False
+
+            comparisons = []
+            for ma_elem, gt_elem in zip(ma_elems, gt_elems):
+                if is_float(gt_elem):
+                    normalized_ma_elem = normalize_number_str(ma_elem)
+                    comparisons.append(normalized_ma_elem == float(gt_elem))
+                else:
+                    ma_elem = normalize_str(ma_elem, remove_punct=False)
+                    gt_elem = normalize_str(gt_elem, remove_punct=False)
+                    comparisons.append(ma_elem == gt_elem)
+            return all(comparisons)
+        else:
+            logger.info(f"Evaluating {model_answer} as a string.")
+            ma_elem = normalize_str(model_answer)
+            gt_elem = normalize_str(ground_truth)
+            return ma_elem == gt_elem
+    except Exception as e:
+        logger.error(f"Error during evaluation: {e}")
+        return False
+
+
+def load_dataset_meta(path: str, split: str = "validation"):
+    data_dir = Path(path) / split
+
+    dataset = []
+    with open(data_dir / "metadata.jsonl", "r", encoding="utf-8") as metaf:
+        lines = metaf.readlines()
+        for line in lines:
+            data = json.loads(line)
+            if data["task_id"] == "0-0-0-0-0":
+                continue
+            if data["file_name"]:
+                data["file_name"] = data_dir / data["file_name"]
+            dataset.append(data)
+    return dataset
+
+
+def load_dataset_meta_dict(path: str, split: str = "validation"):
+    data_dir = Path(path) / split
+
+    dataset = {}
+    with open(data_dir / "metadata.jsonl", "r", encoding="utf-8") as metaf:
+        lines = metaf.readlines()
+        for line in lines:
+            data = json.loads(line)
+            if data["task_id"] == "0-0-0-0-0":
+                continue
+            if data["file_name"]:
+                data["file_name"] = data_dir / data["file_name"]
+            dataset[data["task_id"]] = data
+    return dataset
+
+
+def add_file_path(
+    task: Dict[str, Any], file_path: str = "./gaia_dataset", split: str = "validation"
+):
+    if task["file_name"]:
+        file_path = Path(f"{file_path}/{split}") / task["file_name"]
+        if file_path.suffix in [".pdf", ".docx", ".doc", ".txt"]:
+            task["Question"] += f" Here are the necessary document files: {file_path}"
+
+        elif file_path.suffix in [".jpg", ".jpeg", ".png"]:
+            task["Question"] += f" Here are the necessary image files: {file_path}"
+
+        elif file_path.suffix in [".xlsx", "xls", ".csv"]:
+            task["Question"] += (
+                f" Here are the necessary table files: {file_path}, for processing excel file,"
+                " you can use the excel tool or write python code to process the file"
+                " step-by-step and get the information."
+            )
+        elif file_path.suffix in [".py"]:
+            task["Question"] += f" Here are the necessary python files: {file_path}"
+
+        else:
+            task["Question"] += f" Here are the necessary files: {file_path}"
+
+    return task
+
+
+def report_results(entries):
+    # Initialize counters
+    total_entries = len(entries)
+    total_correct = 0
+
+    # Initialize level statistics
+    level_stats = {}
+
+    # Process each entry
+    for entry in entries:
+        level = entry.get("level")
+        is_correct = entry.get("is_correct", False)
+
+        # Initialize level stats if not already present
+        if level not in level_stats:
+            level_stats[level] = {"total": 0, "correct": 0, "accuracy": 0}
+
+        # Update counters
+        level_stats[level]["total"] += 1
+        if is_correct:
+            total_correct += 1
+            level_stats[level]["correct"] += 1
+
+    # Calculate accuracy for each level
+    for level, stats in level_stats.items():
+        if stats["total"] > 0:
+            stats["accuracy"] = (stats["correct"] / stats["total"]) * 100
+
+    # Print overall statistics with colorful logging
+    logger.info("Overall Statistics:")
+    overall_accuracy = (total_correct / total_entries) * 100
+
+    # Create overall statistics table
+    overall_table = [
+        ["Total Entries", total_entries],
+        ["Total Correct", total_correct],
+        ["Overall Accuracy", f"{overall_accuracy:.2f}%"],
+    ]
+    logger.success(tabulate(overall_table, tablefmt="grid"))
+    logger.info("")
+
+    # Create level statistics table
+    logger.info("Statistics by Level:")
+    level_table = []
+    headers = ["Level", "Total Entries", "Correct Answers", "Accuracy"]
+
+    for level in sorted(level_stats.keys()):
+        stats = level_stats[level]
+        level_table.append(
+            [level, stats["total"], stats["correct"], f"{stats['accuracy']:.2f}%"]
+        )
+
+    logger.success(tabulate(level_table, headers=headers, tablefmt="grid"))
+
+
+import uuid
+import time
+
+from typing import List
+
+import inspect
+from typing import get_type_hints, Tuple
+
+
+def stream_message_template(model: str, message: str):
+    return {
+        "id": f"{model}-{str(uuid.uuid4())}",
+        "object": "chat.completion.chunk",
+        "created": int(time.time()),
+        "model": model,
+        "choices": [
+            {
+                "index": 0,
+                "delta": {"content": message},
+                "logprobs": None,
+                "finish_reason": None,
+            }
+        ],
+    }
+
+
+def get_last_user_message(messages: List[dict]) -> str:
+    for message in reversed(messages):
+        if message["role"] == "user":
+            if isinstance(message["content"], list):
+                for item in message["content"]:
+                    if item["type"] == "text":
+                        return item["text"]
+            return message["content"]
+    return None
+
+
+def get_last_assistant_message(messages: List[dict]) -> str:
+    for message in reversed(messages):
+        if message["role"] == "assistant":
+            if isinstance(message["content"], list):
+                for item in message["content"]:
+                    if item["type"] == "text":
+                        return item["text"]
+            return message["content"]
+    return None
+
+
+def get_system_message(messages: List[dict]) -> dict:
+    for message in messages:
+        if message["role"] == "system":
+            return message
+    return None
+
+
+def remove_system_message(messages: List[dict]) -> List[dict]:
+    return [message for message in messages if message["role"] != "system"]
+
+
+def pop_system_message(messages: List[dict]) -> Tuple[dict, List[dict]]:
+    return get_system_message(messages), remove_system_message(messages)
+
+
+def add_or_update_system_message(content: str, messages: List[dict]) -> List[dict]:
+    """
+    Adds a new system message at the beginning of the messages list
+    or updates the existing system message at the beginning.
+
+    :param msg: The message to be added or appended.
+    :param messages: The list of message dictionaries.
+    :return: The updated list of message dictionaries.
+    """
+
+    if messages and messages[0].get("role") == "system":
+        messages[0]["content"] += f"{content}\n{messages[0]['content']}"
+    else:
+        # Insert at the beginning
+        messages.insert(0, {"role": "system", "content": content})
+
+    return messages
+
+
+def doc_to_dict(docstring):
+    lines = docstring.split("\n")
+    description = lines[1].strip()
+    param_dict = {}
+
+    for line in lines:
+        if ":param" in line:
+            line = line.replace(":param", "").strip()
+            param, desc = line.split(":", 1)
+            param_dict[param.strip()] = desc.strip()
+    ret_dict = {"description": description, "params": param_dict}
+    return ret_dict
+
+
+def get_tools_specs(tools) -> List[dict]:
+    function_list = [
+        {"name": func, "function": getattr(tools, func)}
+        for func in dir(tools)
+        if callable(getattr(tools, func)) and not func.startswith("__")
+    ]
+
+    specs = []
+
+    for function_item in function_list:
+        function_name = function_item["name"]
+        function = function_item["function"]
+
+        function_doc = doc_to_dict(function.__doc__ or function_name)
+        specs.append(
+            {
+                "name": function_name,
+                # TODO: multi-line desc?
+                "description": function_doc.get("description", function_name),
+                "parameters": {
+                    "type": "object",
+                    "properties": {
+                        param_name: {
+                            "type": param_annotation.__name__.lower(),
+                            **(
+                                {
+                                    "enum": (
+                                        param_annotation.__args__
+                                        if hasattr(param_annotation, "__args__")
+                                        else None
+                                    )
+                                }
+                                if hasattr(param_annotation, "__args__")
+                                else {}
+                            ),
+                            "description": function_doc.get("params", {}).get(
+                                param_name, param_name
+                            ),
+                        }
+                        for param_name, param_annotation in get_type_hints(
+                            function
+                        ).items()
+                        if param_name != "return"
+                    },
+                    "required": [
+                        name
+                        for name, param in inspect.signature(
+                            function
+                        ).parameters.items()
+                        if param.default is param.empty
+                    ],
+                },
+            }
+        )
+
+    return specs