fix: prompt and typing

zakahan · zakahan · commit d2c57e64e72d · 2025-05-17T00:11:20.000+08:00
diff --git a/src/agent/analysis_agent/prompt.py b/src/agent/analysis_agent/prompt.py
@@ -1,4 +1,5 @@
-import textwrap  
+import textwrap
+from utils.workflow_type import WorkflowType
 
 AGENT_DESCRIPTION = """
 你是一个热点信息捕捉专家，你现在要做这样的一个任务：
@@ -19,35 +20,34 @@
 3. 招笑瞳姐觉得自己穿过的衣服能卖2w一件🤣【星瞳】
 4. 瞳姐难绷互联网大厂员工不知道工作站电脑开机键在哪【星瞳】
 """
+
+
 # 此处特别致谢恨也迷人，偷了几个标题，用来给LLM参考学习
 
 
-def get_analysis_prompt(query: str, introduction:str, key:str = 'easy'):
-    easy_prompt = textwrap.dedent(
-    f"""
-    ### 你的返回内容应该采取如下的形式
-    ```json
-    [
-        {{
-            "start_time": "来自于输入的start_time字段，表示要切分的视频的开始，而且要讲清楚前因后果",
-            "stop_time": "来自于输入的stop_time字段，表示要切分的的视频的结束"而且要讲清楚前因后果,
-            "title": "表示这段切片视频的标题",
-        }},
-        // 如果你认为这一段录播有多个值得切片的地方，请你继续提出，并且要求前后切片不能有重叠，但你不要切太多，切片的精髓是关键！而不是堆数量！
-    ]
-    ```\n\n""" 
-    )
-
-    two_step_prompt = textwrap.dedent(
-        "暂时我还没想明白咋搞这个，其实前面clip的那个也是我瞎编的，这部分没想好"
-    )
-
-    prompt_dict = {
-        'easy': easy_prompt,
-        'two_step': two_step_prompt
-    }   
-    if key not in prompt_dict:
-        raise KeyError(f"Please set a usable key. I don't know what this '{key}' you found is.")
-
-    back_prompt = f"### 首先给你介绍一下主播的基本信息：\n{introduction}，\n### 切片内容分别如下：\n{str(query)}"
-    return prompt_dict[key] + back_prompt
+def get_analysis_prompt(content: str, introduction: str, key: str = 'easy'):
+    match key:
+        case WorkflowType.EASY:
+            front_prompt = textwrap.dedent(
+                f"""
+                ### 你的返回内容应该采取如下的形式
+                ```json
+                [
+                    {{
+                        "start_time": "来自于输入的start_time字段，表示要切分的视频的开始，而且要讲清楚前因后果",
+                        "stop_time": "来自于输入的stop_time字段，表示要切分的的视频的结束"而且要讲清楚前因后果,
+                        "title": "表示这段切片视频的标题",
+                    }},
+                    // 如果你认为这一段录播有多个值得切片的地方，请你继续提出，并且要求前后切片不能有重叠，但你不要切太多，切片的精髓是关键！而不是堆数量！
+                ]
+                ```\n\n"""
+            )
+        case WorkflowType.TWO_STEP:
+            front_prompt = textwrap.dedent(
+                "暂时我还没想明白咋搞这个，其实前面clip的那个也是我瞎编的，这部分没想好"
+            )
+        case _:
+            raise KeyError(f"Please set a usable key. I don't know what this '{key}' you found is.")
+
+    back_prompt = f"### 首先给你介绍一下主播的基本信息：\n{introduction}，\n### 切片内容分别如下：\n{str(content)}"
+    return front_prompt + back_prompt
diff --git a/src/agent/clip_agent/prompt.py b/src/agent/clip_agent/prompt.py
@@ -1,3 +1,5 @@
+from utils.workflow_type import WorkflowType
+
 AGENT_DESCRIPTION = """
 你是一个视频剪辑专家，你现在要做这样的一个任务：
 现在有一整段视频，来自于一名主播的直播录像，你要担任视频切片员，根据提供的切片指令要求进行切分。
@@ -10,47 +12,41 @@
 """
 
 
-
 def get_clip_prompt(query: dict, key: str) -> str:
-    # 使用 textwrap.dedent 去除公共缩进
-    easy_prompt = (
-        f"你需要对一段视频执行剪辑操作，要求如下\n" + 
-        f"在原素材上切分出一段视频，随后修改标题，要求全过程都使用工具完成。\n" +
-        f"参数如下：\n" +
-        f"1. origin_video_path: {query['origin_video_path']}：\n" +
-        f"2. task_id: {query['task_id']}\n" +
-        f"3. start_time:{query['start_time']}\n" +
-        f"4. stop_time:{query['stop_time']}\n" +
-        f"5. title: {query['title']} \n" +
-        f"请完成剪辑任务，注意，每次任务结束后，都要调用任务结束对应的工具。"
-    )
-
-    two_step_prompt = (
-        f"你需要对一段视频执行剪辑操作，要求如下\n" +
-        f"先在原素材上切分出两段视频，随后按顺序合并。\n" +
-        f"参数如下：\n" +
-        f"1. origin_video_path: {query['origin_video_path']}：\n" +
-        f"2. 第一段视频\n" +
-        f"    2.1 task_id: {query['clip'][0]['task_id']}\n" +
-        f"    2.2. start_time:{query['clip'][0]['start_time']}\n" +
-        f"    2.3. stop_time:{query['clip'][0]['stop_time']}\n" +
-        f"3. 第二段视频\n" +
-        f"    3.1 task_id: {query['clip'][1]['task_id']}\n" +
-        f"    3.2 start_time:{query['clip'][1]['start_time']}\n" +
-        f"    3.3 stop_time:{query['clip'][1]['stop_time']}\n" +
-        f"4. 合并两段视频\n" +
-        f"    4.1 task_id: {query['merge']['task_id']},\n" +
-        f"    4.2 video_paths: 请你根据前两段视频合并操作的结果来决定\n" +
-        f"5. 将视频重命名为\n" +
-        f"请完成剪辑任务\n" 
-    )
-
-
-    prompt_dict = {
-        'easy': easy_prompt,
-        'two_step': two_step_prompt
-    }   
-    if key not in prompt_dict:
-        raise KeyError(f"Please set a usable key. I don't know what this '{key}' you found is.")
-
-    return prompt_dict[key]
+    match key:
+        case WorkflowType.EASY:
+            prompt = (
+                f"你需要对一段视频执行剪辑操作，要求如下\n" +
+                f"在原素材上切分出一段视频，随后修改标题，要求全过程都使用工具完成。\n" +
+                f"参数如下：\n" +
+                f"1. origin_video_path: {query['origin_video_path']}：\n" +
+                f"2. task_id: {query['task_id']}\n" +
+                f"3. start_time:{query['start_time']}\n" +
+                f"4. stop_time:{query['stop_time']}\n" +
+                f"5. title: {query['title']} \n" +
+                f"请完成剪辑任务，注意，每次任务结束后，都要调用任务结束对应的工具。"
+            )
+        case WorkflowType.TWO_STEP:
+            prompt = (
+                f"你需要对一段视频执行剪辑操作，要求如下\n" +
+                f"先在原素材上切分出两段视频，随后按顺序合并。\n" +
+                f"参数如下：\n" +
+                f"1. origin_video_path: {query['origin_video_path']}：\n" +
+                f"2. 第一段视频\n" +
+                f"    2.1 task_id: {query['clip'][0]['task_id']}\n" +
+                f"    2.2. start_time:{query['clip'][0]['start_time']}\n" +
+                f"    2.3. stop_time:{query['clip'][0]['stop_time']}\n" +
+                f"3. 第二段视频\n" +
+                f"    3.1 task_id: {query['clip'][1]['task_id']}\n" +
+                f"    3.2 start_time:{query['clip'][1]['start_time']}\n" +
+                f"    3.3 stop_time:{query['clip'][1]['stop_time']}\n" +
+                f"4. 合并两段视频\n" +
+                f"    4.1 task_id: {query['merge']['task_id']},\n" +
+                f"    4.2 video_paths: 请你根据前两段视频合并操作的结果来决定\n" +
+                f"5. 将视频重命名为\n" +
+                f"请完成剪辑任务\n"
+            )
+        case _:
+            raise KeyError(f"Please set a usable key. I don't know what this '{key}' you found is.")
+
+    return prompt
diff --git a/src/main.py b/src/main.py
@@ -1,5 +1,6 @@
 import asyncio
 import os
+import secrets
 from dotenv import load_dotenv
 # 加载.env文件中的环境变量
 load_dotenv()
@@ -11,8 +12,8 @@
 introduction= json2dict("introduction.json")
 
 query = {
-        "task_id":"007",
-        "raw_video":"raw/【直播回放】不动如山 2025年01月06日20点场.mp4",
+        "task_id":secrets.token_hex(4),
+        "raw_video":"raw/test.mp4",
         "introduction":introduction["雫るる_Official"]
 }
 
diff --git a/src/processor/analysis_processor.py b/src/processor/analysis_processor.py
@@ -25,14 +25,17 @@ def __init__(self):
         
         pass
 
-    async def run(self, query: str, introduction:str, prompt_key: str='easy') -> str:
+    async def run(self, query: dict, prompt_key: str='easy') -> str:
+        asr_content = query['content']
+        introduction = query['introduction']
+
         agent = get_analysis_agent()
         runner = Runner(
             app_name=self.app_name,
             agent=agent, 
             session_service=self.session_service
         )
-        prompt=get_analysis_prompt(query, introduction, prompt_key)
+        prompt=get_analysis_prompt(asr_content, introduction, prompt_key)
         content = types.Content(role='user', parts=[types.Part(text=prompt)])
         events_async = runner.run_async(
             session_id=self.session_id,
diff --git a/src/processor/root_processor.py b/src/processor/root_processor.py
@@ -34,8 +34,6 @@ def _get_video_path(self, task_id_list: list[str]) -> list[dict]:
             dir_path = os.path.join(os.getenv("KB_BASE_PATH"), "result", task_id)
             v_list = find_video_files(dir_path)
             res_list.extend(v_list)
-        
-        
         return res_list
 
 
@@ -52,14 +50,21 @@ async def run(self, query: dict) -> dict:
             query={
                 "input_audio": raw_video,
                 "task_id": asr_task_id
-        })
+            }
+        )
         
         # 分析阶段 ---------------------------------------------------------
         aly_pcr = AnalysisProcessor()
-        aly_task_id_group = [f"{task_id}_aly_{i}" for i in range(len(asr_results['batch']))]
+        # aly_task_id_group = [f"{task_id}_aly_{i}" for i in range(len(asr_results['batch']))]
         aly_results = []
-        for i, aly_task_id in enumerate(aly_task_id_group):
-            aly = await aly_pcr.run(asr_results['batch'][i], introduction)
+        for i in range(len(asr_results['batch'])):
+            aly = await aly_pcr.run(
+                query={
+                    'content': str(asr_results['batch'][i]),
+                    'introduction': introduction
+                },
+                prompt_key=self.prompt_key
+            )
             aly_results.append(aly)
             pass
 
diff --git a/src/utils/__init__.py b/src/utils/__init__.py
diff --git a/src/utils/workflow_type.py b/src/utils/workflow_type.py
@@ -0,0 +1,8 @@
+from enum import StrEnum
+
+# must python 3.11 +
+class WorkflowType(StrEnum):
+    EASY = 'easy'
+    TWO_STEP = 'two_step'
+
+
