agent.py

from model import DialogueAgent, simple_llm
import json
import re
import threading
import time
import keyboard
from retrieval_data import get_retrival_data, get_retrieval_data_from_instruction,get_retrival_risk_data_from_subtask
from dcs_export_screen import analyze_dcs_screen
from rag_test import rag
from planner_global import TaskPlanner

import queue

# generate_instruction_and_check_func prompt
tmp1 = """
You are the F-14 fighter jet assistance system. Please help the pilot complete the mission according to the following rules:

Next, you need to guide the pilot step by step based on the information above. Please generate guidance information for the pilot based on the previous information (example action sequence, relevant data names, current data).
Note that you need to determine whether the current step needs to be appropriately modified based on the current information. Please output the step you consider appropriate.

And since the pilot may not be aware of the location of the required switches or instruments, you need to provide their positions based on the data.

In addition, when generating specific operations, you also need to generate a judgment function to determine whether the current step is completed, which will be used for subsequent step planning.
This function should be in the form of Python code, and the data involved in the function should be selected from the previously retrieved actual data. Be careful not to arbitrarily generate data that may not exist.
The format reference is as follows:
example:

## normal case
input:
<step info>
Set the MSL PREP switch ON. This commands the WCS to start missile preparation for the AIM-7 and AIM-54. When the individual missiles are tuned and ready the corresponding missile status windows turns white to indicate a ready missile. This should take approx. 2 minutes.
</step info>

<specific data>
name: PLT_MISSLE_PREP
description: Initiates missile preparation sequence for AIM-7 and AIM-54 weapon systems.
api_variant: momentary_last_position
value_scale: [\'0\':\'OFF\', \'1\':\'ON\']
position: ACM Panel

name: PLT_MSL_PREP_ON
description: Indicates missile preparation status for pilot awareness.
api_variant: None
value_scale: [\'0\':\'OFF\', \'1\':\'ON\']
position: ACM Panel
</specific data>

output:
<think>
this step need to Set the MSL PREP switch ON, it located on ACM Panel so I should info pilot to operate it and tell the position.
</think>

<text>
Set the MSL PREP switch ON, it located in ACM Pannel.
</text>

<check func>
result = dcs_states['PLT_MSL_PREP_ON'] == "ON"
</check func>

## multi-options case
for some multi-options step, you can use more complex func to detect the status, for example:


## screen type case
When processing api_variant data of screen type, follow these guidelines:
1. For screen-type information (visual interface data), generate string-based queries that will be co-processed with the screen capture by a Vision-Language Model (VLM). 

2. Format requirements:
   - Use natural language questions in quotation marks
   - Ensure questions target specific visual elements
   - Maintain aircraft system terminology consistency

NOTICE: When no specific screen type data is provided, absolutely do not attempt to access screens that may not exist without proper evidence. In such cases, prioritize using non-screen data whenever possible.
   
for example:
the instruction is:
"Line up target with ADL (within 20 deg) in HUD"
and the related data is not able to represent ADL, so you must analyze the HUD screen, so your check function is:

result = self.analysis_screen(screen_name="F_14_HUD", prompt="Does the target have already been line up with ADL(within 20 deg)?")

## format note:
Note: Please output the thinking, the step instructions and check functions strictly in the following format, and do not output anything else:
- Thinking shoud be contained in <think>...</think> tags.
    you may need choose some data you want use in check function generating, and think what pilot need to know, but your thinking will not show to pilot.

- Step instructions should be contained in <text>...</text> tags.
For step instructions, Please omit some descriptive statements and keep it concise, using the fastest speed to make the user understand the current step.

- Check functions should be contained in <check func>...</check func> tags.
For check function:
- never add comment in the check function, which may result to syntax error.
- dcs_states is a dictionary, ensure don't use dcs_states['xx']['xx'] in the check function, instead use dcs_states['xx'] directly.
- self.analysis_screen is a function in class, you can directly use it in the check function in correct format.
- If the value scale is given(inf you the number's meaning), please use the number's meaning instead of using the number;
for example:
name: "PLT_HUD_MODE_AWL"
value_scale:{"1": "ILS","0": "ACL"}
if the value is 1, it means the value is ILS,check function should be:
result = dcs_states['PLT_HUD_MODE_AWL'] == "ILS"

- If there is no value scale given, you can use 0 or 1 (type:int) as the judgment value.
"""

#Deprecated
tmp2 = """
你是F14战斗机的辅助驾驶程序，你需要预判一些潜在的驾驶危险并生成一些检查函数在后台运行，当突发情况发生时可以立即提供驾驶员反馈信息。

##Example:
<subtask>
AIM-7M SPARROW LAUNCH PROCEDURE WITH RADAR TRACKING
</subtask>

<risk>
"discription":
"data":
</risk>
<check_func>
if data == True:
    result = ""
</check_func>

<risk>

</risk>
<check_func>

</check_func>

<risk>

</risk>
<check_func>

</check_func>

"""

tmp3 = """
<step info>
{step_num}:{step_ins}
</step info>

<common data>
{common_data}
</common data>

<specific data>
{specific_data}
</specific data>
Based on the above information, please provide the user with a language prompt and a check function.

Please remember that some unnecessary data may be provided to you, such as redundant information, irrelevant information, etc.

Please combine the context and actual scenario and use only the required data in the check function.

If the api_variant of a data is not None, please pay special attention to it.

For example, momentary_last_position means that this key is a sequential key, which is suitable for detecting whether the key is pressed and should not be relied on to determine the system status.

if feedback is not empty, please be sure to pay attention to the help provided in the feedback
<feedback>
{feedback}
</feedback>
"""

# central agent prompt
tmp4 = """
You are an F-14 aircraft assistance system, and your task is to help the pilot complete missions. The current task is {subtask_name}, and you need to use various tools to help the pilot complete the task, including but not limited to: providing guidance information, checking task execution status, analyzing screen information, etc.

You need to call generate_task_plan() once to generate a task plan for the current mission (Note: you only need to use generate_task_plan() once in this conversation)

After that, you should call the next_step function to get the next step instruction first.
Then you should call the get_relevant_data function to get the relevant data.
finally, call the generate_instruction_and_check_func function to generate the instruction and check function to detect whether the user finsished action.
After that, if you are reminded that the current action has been completed by the user, call next_step() to continue the loop.

Please strictly follow the above steps,and refrain from invoking any additional functions unless the user explicitly requests or requires them.

If there is external input information from the user, please call tools or respond to the user's questions.

Please ensure your output is concise and does not include any irrelevant information. when you need to use tools, only output tool calls, do not output any other information.

<NOTICE: you can only use one tool at a time.>
"""


##agent除了开放调用的tool，还有一些只支持内部调用的tool，例如analys screen，只会在生成代码阶段进行调用
##提出了内部工具的概念
class CopilotAgent():
    def __init__(self, task_name,agent_name, dcs_state, embeddings, model_dict, logger=None, history=None, plan = [], finished_list=[],eval_mode=False):
        self.task_name = task_name
        self.logger = logger
        if self.logger:
            self.logger.info(f"###Task name: {self.task_name}")
        self.last_context_length = 0#辅助logger记录上次的上下文长度

        self.dcs_state = dcs_state
        self.embeddings = embeddings
        self.model_dict = model_dict
        self.agent = DialogueAgent(self.model_dict["agent"], tools=tools, system_msg=tmp4)
        self.agent_name = agent_name
        # self.plan_agent = TaskPlanner(action_path='data\json\subtasksV3.json',
        #                                 task = self.task_name,
        #                                 model=self.model_dict["planner"],
        #                                 init_obs=self.init_state)

        self.finished_list = finished_list  # 用于记录动作是否完成的列表
        self.eval_mode = eval_mode  # 是否为评估模式，评估模式下不执行实际操作，只生成指令和检查函数
        self.step_count = 0
        self.instruction = ""
        self.step_info = []#step_info是当前需要执行的动作相关的信息，不一定是plan中的step
        self.check_thread = None
        self.check_func_finished_event = None
        self.message_queue = queue.Queue()  # 用于存储外部消息
        self.is_running = True  # 控制主循环
        self.is_waiting_for_check = False  # 是否正在等待检查函数完成

        self.plan = plan
        self.instruction_counter = 0  # 用于生成唯一的指令ID
        self.monitoring_data = {}  # 存储所有需要监控的数据 {instruction_id: {"names": [], "check_func": "", "event": Event, "instruction": ""}}
        self.check_threads = {}    # 存储所有监控线程 {instruction_id: Thread}

    # 调用后返回查询相关的文档信息辅助回答
    def get_relevant_docs(self, query):
        # unimplemented
        tmp = """你是F14战斗机辅助系统，需要帮助用户回答问题{query}
        
        请根据以下检索到的信息以及你的专业知识回答问题：
        {relevant_docs}
        """
        output = rag(query, self.embeddings, tmp, slient=False)
        return output

    # 基于当前指令，检索当前指令最相关的数据信息，并将其保存在self.step_info中
    def get_relevant_data(self, feedback=None):
        # param: ins:当前指令
        #       feedback:反馈信息，如果存在，则需要基于反馈信息进一步细化检索
        #解锁
        self.is_waiting_for_check = False

        self.step_info = get_retrieval_data_from_instruction(
                instruction=self.agent_name + ":" + self.instruction,
                embeddings=self.embeddings,
                model_name=self.model_dict["retriever"],
                with_mapping="data\json\dcs\mapping\F-14_mapping.json",
                slient=False,
                logger=self.logger,
                feedback=feedback
            )
        return "Successfully retrieved relevant data and stored in self.step_info, you can directly execute other functions, and they will automatically use these data"
    
    def correct_check_function_code(self, check_func_code):
        """
        纠正LLM生成的检查函数代码中的常见错误，並清除換行符號
        
        Args:
            check_func_code (str): 原始检查函数代码
        
        Returns:
            str: 纠正后的检查函数代码
        """
        # 修復重複鍵名訪問錯誤（如 dcs_states['PLT_HUD_MODE']['PLT_HUD_MODE']）
        pattern = r"dcs_states\['([^']+)'\]\['(\1)'\]"
        corrected_code = re.sub(pattern, r"dcs_states['\1']", check_func_code)
        
        # 清除所有換行符號
        corrected_code = corrected_code.replace('\n', '').replace('\r', '')
        corrected_code = re.sub(r'\s+', ' ', corrected_code)
        # 如果有修改，打印日志
        if corrected_code != check_func_code:
            print("纠正了检查函数中的重复键名错误或清除了换行符")
        
        return corrected_code

    def _is_fully_enclosed(self,expr):
        """檢查表達式是否被一對從頭到尾的括號完全包裹"""
        expr = expr.strip()
        if not (expr.startswith('(') and expr.endswith(')')):
            return False
        
        # 檢查括號是否匹配且中間沒有閉合
        paren_count = 0
        # 從第二個字符遍歷到倒數第二個字符
        for char in expr[1:-1]:
            if char == '(':
                paren_count += 1
            elif char == ')':
                paren_count -= 1
            # 如果計數器在遍歷結束前降到負數，說明括號不匹配
            if paren_count < 0:
                return False
        return True

    # 重寫：使用工作佇列的迭代替代方案，以處理嵌套邏輯
    def parse_and_split_check_functions(self, check_func_code):
        """
        解析並拆分檢查函式，能正確處理複雜的括號嵌套。
        """
        check_func_code = check_func_code.strip()
        
        # 使用 re.DOTALL 以匹配可能跨行的表達式
        result_pattern = r'result\s*=\s*(.+)'
        result_match = re.search(result_pattern, check_func_code, re.IGNORECASE | re.DOTALL)
        
        if not result_match:
            # 如果沒有找到 `result =`，則返回原始代碼
            if ' and ' not in check_func_code.lower() and ' or ' not in check_func_code.lower():
                return [f"result = {check_func_code}"]
            return [check_func_code]
            
        expression = result_match.group(1).strip()

        # 如果表達式中沒有 'and'，則直接返回，無需拆分
        if ' and ' not in expression.lower():
            return [f"result = {expression}"]

        final_parts = []
        processing_queue = [expression] # 初始化工作佇列

        while processing_queue:
            current_expr = processing_queue.pop(0)

            # 迭代地移除包裹整個表達式的外層括號
            while self._is_fully_enclosed(current_expr):
                current_expr = current_expr[1:-1].strip()

            # 使用現有的分割函式在頂層進行分割
            parts = self._split_by_operator(current_expr, ' and ')

            if len(parts) <= 1:
                # 如果不能再分割（或只有一部分），則將其添加到最終列表中
                # 這部分可能是簡單條件，或是一個 `or` 表達式塊
                final_parts.append(current_expr)
            else:
                # 如果成功分割，將新的部分添加回佇列進行下一輪處理
                processing_queue.extend(parts)

        # 格式化輸出
        check_functions = [f"result = {part}" for part in final_parts]
        return check_functions
    
    def _split_by_operator(self, expression, operator):
        """
        根據操作符分割表達式，但保持括號內的完整性
        
        Args:
            expression (str): 要分割的表達式
            operator (str): 操作符 (' and ' 或 ' or ')
        
        Returns:
            list: 分割後的部分列表
        """
        parts = []
        current_part = ""
        paren_count = 0
        
        # 將表達式轉換為小寫進行比較
        expression_lower = expression.lower()
        operator_lower = operator.lower()
        
        i = 0
        while i < len(expression):
            char = expression[i]
            
            if char == '(':
                paren_count += 1
            elif char == ')':
                paren_count -= 1
            
            # 檢查是否匹配操作符（不區分大小寫）
            if paren_count == 0 and i + len(operator) <= len(expression):
                potential_operator = expression[i:i+len(operator)]
                if potential_operator.lower() == operator_lower:
                    if current_part.strip():
                        parts.append(current_part.strip())
                    current_part = ""
                    i += len(operator)
                    continue
            
            current_part += char
            i += 1
        
        # 添加最後一部分
        if current_part.strip():
            parts.append(current_part.strip())
        
        return parts

    # 重点函数，用于生成指导信息和检查函数
    def generate_instruction_and_check_func(self,feedback=None):
        print("#############################")
        specific_data_str = ""
        
        # 生成唯一的指令ID
        instruction_id = f"instruction_{self.instruction_counter}"
        self.instruction_counter += 1
        
        # 初始化该指令的监控数据
        self.monitoring_data[instruction_id] = {
            "names": [],
            "check_func": None,
            "event": None,
            "instruction": self.instruction  # 存储指令内容
        }
        
        if self.step_info == []:
            return f"you should call get_relevant_data first"
        
        zip_step_info = [j for i in self.step_info for j in i["keys"]]

        for data in zip_step_info:
            self.monitoring_data[instruction_id]["names"].append(data['key'])
            # 从mapping中获取可能的值
            value_scale = data.get("value_scale", {})
            api_variant = data.get("api_variant", None)
            position = data.get("position",None)
            specific_data_str += f"***\nname: {data['key']}\ndescription: {data['description']}\nvalue: {self.dcs_state.get_chosen_data([data['key']])}\n value_scale: {value_scale}\napi_variant:{api_variant}\nposition: {position}\n***\n"


        llm_output = tmp3.format(
            step_num=self.step_count,
            step_ins=self.instruction,
            common_data=self.dcs_state.get_common_data(),
            specific_data=specific_data_str,
            feedback=feedback
        )
        
        output = simple_llm(name=self.model_dict["check"], content=llm_output, sys=tmp1)

        ## 解析输出
        text_pattern = r'<text>(.*?)</text>'
        check_pattern = r'<check func>(.*?)</check func>'
        text_match = re.search(text_pattern, output, re.DOTALL)
        check_match = re.search(check_pattern, output, re.DOTALL)
        
        if text_match:
            text = text_match.group(1).strip()
            # print(f"instruction match:\n {text}")

        if check_match:
            check_func_code = check_match.group(1).strip()
            check_func_code = self.correct_check_function_code(check_func_code)
            
            # 解析和拆分檢查函數
            split_check_functions = self.parse_and_split_check_functions(check_func_code)
            print(check_func_code)
            print(split_check_functions)
            # 為每個拆分後的檢查函數創建獨立的監控
            check_func_ids = []
            for i, split_func in enumerate(split_check_functions):
                split_instruction_id = f"{instruction_id}_check_{i}"
                check_func_ids.append(split_instruction_id)
                
                # 初始化該檢查函數的監控數據
                self.monitoring_data[split_instruction_id] = {
                    "names": self.monitoring_data[instruction_id]["names"].copy(),
                    "check_func": split_func,
                    "event": threading.Event(),
                    "instruction": self.instruction,
                    "parent_instruction_id": instruction_id,
                    "completed": False
                }
                
                # 啟動監控線程
                self.check_threads[split_instruction_id] = threading.Thread(
                    target=self.monitor_check_function,
                    args=(split_instruction_id,)
                )
                self.check_threads[split_instruction_id].daemon = True
                self.check_threads[split_instruction_id].start()
            
            # 更新主指令的監控數據，包含所有子檢查函數的ID
            self.monitoring_data[instruction_id]["check_func_ids"] = check_func_ids
            self.monitoring_data[instruction_id]["event"] = threading.Event()
            
            self.is_waiting_for_check = True
                
        return f"Successfully generated instruction:\n {text} and check function:\n{check_func_code}\nwaiting for check functions to complete."

    # 调用后读取下一条指导操作
    def next_step(self):
        # 停止所有正在运行的监控线程
        active_instruction_ids = list(self.monitoring_data.keys())
        for instruction_id in active_instruction_ids:
            monitor_info = self.monitoring_data[instruction_id]
            if monitor_info["event"]:
                monitor_info["event"].set()
            if instruction_id in self.check_threads:
                self.check_threads[instruction_id].join(timeout=1)
                del self.check_threads[instruction_id]
            del self.monitoring_data[instruction_id]
        
        #解锁
        self.is_waiting_for_check = False

        self.step_info = []
        # if self.plan == []:
        #     return "You need call generate_task_plan() first to get plan."
        # 读取下一条指令
        if self.step_count < len(self.plan): 
            self.instruction = self.plan[self.step_count].get("instruction","")#step应该从0开始记录
        else:
            return "Current task is completed"
        
        self.step_count += 1
        return f"Next step instruction: {self.instruction}, you can call get_relevant_data to get the relevant data"
    
    # 执行检查函数
    def execute_check_function(self, instruction_id):
        # return True#测试模式，默认每条指令都执行成功
        if instruction_id not in self.monitoring_data:
            return False
            
        monitor_info = self.monitoring_data[instruction_id]
        global_vars = {
            "self": self,
            "__builtins__": __builtins__
        }
        local_vars = {
            'dcs_states': self.dcs_state.get_chosen_data(monitor_info["names"])
        }
        if self.eval_mode:
            return True
        exec(monitor_info["check_func"], global_vars, local_vars)
        # return True#测试模式，默认每条指令都执行成功
        return local_vars.get('result', False)
    #设立两种监督函数：
    # 一种作为判断程序是否继续执行，对于用于判断程序是否继续执行的监督函数，返回值是True/False，
    # 一种处理用户需求和risk，返回带信息的字符串，用于和用户进行交互
    # 监控检查函数
    def monitor_check_function(self, instruction_id):
        time.sleep(0.1)
        print(f"monitor_check_function start for {instruction_id}")
        
        monitor_info = self.monitoring_data[instruction_id]
        
        # 檢查是否為子檢查函數
        if "parent_instruction_id" in monitor_info:
            # 這是子檢查函數，只監控自己的條件
            while not monitor_info["event"].is_set():
                time.sleep(0.1)
                if self.execute_check_function(instruction_id):
                    print(f"Sub-condition met for {instruction_id}!")
                    monitor_info["completed"] = True
                    monitor_info["event"].set()
                    
                    # 檢查父指令的所有子檢查函數是否都完成
                    parent_id = monitor_info["parent_instruction_id"]
                    if parent_id in self.monitoring_data:
                        parent_info = self.monitoring_data[parent_id]
                        if "check_func_ids" in parent_info:
                            all_completed = True
                            for check_id in parent_info["check_func_ids"]:
                                if check_id in self.monitoring_data:
                                    if not self.monitoring_data[check_id]["completed"]:
                                        all_completed = False
                                        break
                            
                            if all_completed:
                                print(f"All sub-conditions completed for {parent_id}!")
                                # 觸發父指令的完成邏輯
                                self._handle_instruction_completion(parent_id)
                    break
        else:
            # 這是主檢查函數（沒有拆分的單一檢查函數）
            while not monitor_info["event"].is_set():
                time.sleep(0.1)
                if self.execute_check_function(instruction_id):
                    print(f"Condition met for {instruction_id}!")
                    self._handle_instruction_completion(instruction_id)
                    break
    
    def _handle_instruction_completion(self, instruction_id):
        """
        處理指令完成時的邏輯
        """
        monitor_info = self.monitoring_data[instruction_id]
        print(self.instruction)
        
        # --- **修改点：增加 eval_mode 判断** ---
        # 只有在非评估模式（即真实运行时），才执行与 plan 和 finished_list 相关的逻辑
        if not self.eval_mode:
            # 建立多agent之間操作的依賴關係
            self.finished_list[self.plan[self.step_count - 1]['step'] - 1] = True

            #建立多机通讯
            if hasattr(self, 'host_manager') and self.host_manager is not None:
                self.host_manager.push_finished_list(self.finished_list)

            # 檢查是否還有下一步，如果有才檢查依賴關係
            if self.step_count < len(self.plan):
                dependency = self.plan[self.step_count].get("dependency", None)
                print(dependency)
                if dependency is not None and self.finished_list[dependency - 1] is False:
                    print(f"Waiting for other one complete before proceeding.")
                    print(self.finished_list)
                    while not self.finished_list[dependency - 1]:
                        time.sleep(0.1)
            else:
                print("No more steps in plan, task completed.")
        
        completion_message = f"Current step <{monitor_info['instruction']}> has been confirmed to be completed, please continue to the next step."
        self.agent.update_context(completion_message, role="assistant")
        
        # 设置事件
        monitor_info["event"].set()
        self.is_waiting_for_check = False

    # 新增：支持外部设置 finished_list（用于客户端接收主机推送）
    def set_finished_list(self, finished_list):
        self.finished_list = finished_list

    # 基于prompt调用外部VLM分析指定显示器的信息，将分析结果返回##修改，bool输出不再有意义，因为bool用法不可能开放给LLM直接调用，都是其他函数间接调用
    # 由于是其他函数调用，因此screen name可以保证正确
    def analysis_screen(self, screen_name=None, prompt=None):
        # param: screen_name:屏幕名称
        #       prompt:分析提示
        #analysis_screen不应该对外开放，只是开放给generate_instruction_and_check_func使用，如果对外开放，无法解决自主判断何时调用的问题
        #例如：用户提问需要分析HUD屏幕，但agent并不知道需要

        prompt = f"{prompt}\n<Please output the result in bool type, only return True or False, do not include any other information>"
        screen_prompt_dict = {
            screen_name: prompt
        }
        result = analyze_dcs_screen(screen_prompt_dict, slient=False,logger=self.logger)
        return result
    
    #开放函数，用于解决用户提出的视频相关问题
    def query_screen(self, prompt=None):
        #从step info中筛选出screen_name对应的api_variant数据
        if self.step_info == []:
            return f"you need call get_relevant_data first"
        
        screen = []
        zip_step_info = [j for i in self.step_info for j in i["keys"]]

        for data in zip_step_info:
            if data["api_variant"] == "Screen":
                screen.append(data["key"])
        screen_prompt_dict = {
            screen[0]:prompt
        }
        ###暂时只考虑映射到一个显示屏的情况，不考虑其他情况
        result = analyze_dcs_screen(screen_prompt_dict, slient=False,logger=self.logger)
        return result

    #生成当前用户需求下的整体任务，每次启动前都必须调用
    def generate_task_plan(self):
        # if self.plan != []:
        #     return "Plan already exsist, you should call regenerate_task_plan to re-plan."
        # self.plan = self.plan_agent.fastplan()
        if self.logger:
            self.logger.info(f"plan: {self.plan}")
        return "The action sequence for the current task has been generated in self.plan. Next, use next_step() to get the action instructions to be executed."
    
    def regenerate_task_plan(self,query):
        #针对用户提出的新需求，重新对原本的任务动作序列进行规划
        if self.plan == []:
            return "No existing task plan, please use generate_task_plan to generate the task plan for the first time"
        
        #抽取当前任务完成情况
        history = ""


        # self.plan = self.plan_agent.replan(history=history,query=query)

        return "The action sequence for the new requirement has been replanned. Next, use next_step() to get the action instructions to be executed."

    ## 外部程序接口
    # 添加外部消息到队列
    def add_message(self, message):
        tmp = """maybe there still have some check function are running, but User send a message, you need help user to solve the problem, so you need to add the message to the queue first"""
        self.message_queue.put(message)
    
    # 获取当前对话内容
    def get_chat_content(self):
        return self.agent.get_chat_content()
    
    # 清空对话内容
    def clear_chat_content(self, clear_sys=True):
        self.agent.clear_chat_content(clear_sys)
    
    # 停止运行
    def stop(self):
        self.is_running = False
        # 停止所有监控线程
        for monitor_info in self.monitoring_data.values():
            if monitor_info["event"]:
                monitor_info["event"].set()
    
    # 主循环，处理消息和执行agent
    def mainloop(self):
        self.agent.update_context(message=f"Please start the task, {self.task_name}", role="user")
        while self.is_running:
            # 检查是否有外部消息
            try:
                message = self.message_queue.get_nowait()
                # 处理外部消息
                #更新self.instrution为用户指令
                self.instruction = message
                self.agent.update_context(message=message, role="user")
                response = self.agent.generate_response()
                self.message_queue.task_done()
            except queue.Empty:
                if not self.is_waiting_for_check:
                    print("###send request###")
                    # print(self.agent.context)
                    response = self.agent.generate_response()
            if response is not None:
                print("###response###")
                print(response)

            # 处理工具调用
            if hasattr(response, 'tool_calls') and response.tool_calls:
                for tool_call in response.tool_calls:
                    try:
                        function_name = tool_call.function.name
                        # 添加参数检查和错误处理
                        try:
                            if not tool_call.function.arguments or tool_call.function.arguments.strip() == "":
                                arguments = {}
                            else:
                                arguments = json.loads(tool_call.function.arguments)
                        except json.JSONDecodeError as e:
                            print(f"JSON解析错误: {e}")
                            print(f"原始参数字符串: {tool_call.function.arguments}")
                            arguments = {}

                        ###debug
                        print(f"执行工具调用：{function_name}")
                        print(f"执行工具调用参数：{arguments}")

                        # 执行相应的函数
                        if function_name == "get_relevant_docs":
                            result = self.get_relevant_docs(arguments.get("query", ""))
                        elif function_name == "get_relevant_data":
                            result = self.get_relevant_data(
                                arguments.get("feedback", None)
                            )
                        elif function_name == "generate_instruction_and_check_func":
                            result = self.generate_instruction_and_check_func()
                        elif function_name == "next_step":
                            result = self.next_step()
                        elif function_name == "query_screen":
                            result = self.query_screen(
                                arguments.get("prompt", "")
                            )
                        elif function_name == "generate_task_plan":
                            result = self.generate_task_plan()
                        # 将结果添加到对话中
                        print(f"{result}")
                        self.agent.update_context(message=result, role="tool", tool_call_id=tool_call.id)
                    except Exception as e:
                        error_msg = f"function call error: {str(e)}"
                        print(error_msg)
                        self.agent.update_context(message=error_msg, role="tool", tool_call_id=tool_call.id)
            
            ###每次变化长度时，都需要记录下消息变化
            if len(self.agent.context) != self.last_context_length:
                for message in self.agent.context[self.last_context_length:]:
                    if isinstance(message, dict):
                        role = message["role"]
                        content = message["content"]
                    else:
                        role = message.role
                        content = message.content
                    # 处理助手消息中可能包含的工具调用
                    if role == "assistant" and hasattr(message, "tool_calls") and message.tool_calls:
                        # 由于限定agent一次只会调用一个tool，所以只取第一个tool_call
                        tool_call = message.tool_calls[0]
                        if self.logger:
                            self.logger.info(f"function call: {tool_call.function.name}({tool_call.function.arguments})")
                    else:
                        if self.logger:
                            self.logger.info(f"{role}: {content}")
                        else:
                            pass
                self.last_context_length = len(self.agent.context)
            if result == "Current task is completed":
                self.is_running = False
                break
            response = None  # 清空response,避免重复调用
            time.sleep(0.1)  # 避免CPU占用过高

    def run(self):
        """Run the main loop in a separate thread."""
        self.check_thread = threading.Thread(target=self.mainloop)
        self.check_thread.daemon = True
        self.check_thread.start()

tools = [
    {
        "type": "function",
        "function": {
            "name": "get_relevant_docs",
            "description": "get relevant docs from the database",
            "parameters": {
                "type": "object",
                "properties": {
                    "query": {
                        "type": "string",
                        "description": "query string, used to retrieve relevant doc information"
                    }
                },
                "required": ["query"]
            }
        }
    },
    {
        "type": "function",
        "function": {
            "name": "get_relevant_data",
            "description": "find some data relevant to the current instruction in the database",
            "parameters": {
                "type": "object",
                "properties": {
                    "feedback": {
                        "type": "string",
                        "description": "feedback information, if exists, it needs to be further refined based on the feedback information.",
                        "nullable": True
                    }
                },
                "required": ["ins"]
            }
        }
    },
    {
        "type": "function",
        "function": {
            "name": "generate_instruction_and_check_func",
            "description": "generate instruction for User and check function to check if the current step has been completed, it will use analysis_screen to analyze the screen information if needed",
            "parameters": {
                "type": "object",
                "properties": {},
                "required": []
            }
        }
    },
    {
        "type": "function",
        "function": {
            "name": "next_step",
            "description": "read the next instruction, You should call it in first time or when the current step has been completed",
            "parameters": {
                "type": "object",
                "properties": {},
                "required": []
            }
        }
    },
    {
        "type": "function",
        "function": {
            "name": "query_screen",
            "description": "analyze the information of the aircraft displays based on the relevant data and user query. Use this tool when user asks questions about what's shown on the aircraft screens",
            "parameters": {
                "type": "object",
                "properties": {
                    "prompt": {
                        "type": "string",
                        "description": "specific question about what to look for on the screen"
                    }
                },
                "required": ["prompt"]
            }
        }
    },
    {
        "type": "function",
        "function": {
            "name": "generate_task_plan",
            "description": "Generate the overall task plan based on the current user's requirements. This function must be called before each start.",
            "parameters": {
                "type": "object",
                "properties": {},
                "required": []
            }
        }
    }
]