feat: parallel_check

Author: lilingfengdev

api/v1/chat.py

@@ -160,6 +160,7 @@ def on_agent_update(agent_id: str, update: Dict[str, Any]):
             model_stages=model_config.models,
             on_progress=on_progress,
             on_agent_update=on_agent_update,
+            enable_parallel_check=model_config.parallel_check,
             llm_params=llm_params,
         )
 
@@ -242,6 +243,7 @@ def on_progress(event):
             model_stages=model_config.models,
             on_progress=on_progress,
             enable_planning=model_config.has_plan_mode,
+            enable_parallel_check=model_config.parallel_check,
             llm_params=llm_params,
         )
 
@@ -404,6 +406,7 @@ async def chat_completions(
                 num_agents=model_config.num_agent,
                 parallel_run_agent=model_config.parallel_run_agent,
                 model_stages=model_config.models,
+                enable_parallel_check=model_config.parallel_check,
                 llm_params=llm_params,
             )
             result = await engine.run()
@@ -423,6 +426,7 @@ async def chat_completions(
                 required_successful_verifications=model_config.required_verifications,
                 model_stages=model_config.models,
                 enable_planning=model_config.has_plan_mode,
+                enable_parallel_check=model_config.parallel_check,
                 llm_params=llm_params,
             )
             result = await engine.run()

config.py

@@ -21,6 +21,7 @@ def __init__(self, model_id: str, config: Dict[str, Any]):
         self.max_iterations = config.get("max_iterations", 30)
         self.required_verifications = config.get("required_verifications", 3)
         self.max_errors = config.get("max_errors_before_give_up", 10)
+        self.parallel_check = config.get("parallel_check", False)  # 并行验证模式
 
         # UltraThink 配置
         self.num_agent = config.get("num_agent")

config.yaml.example

@@ -37,6 +37,7 @@ model:
     rpm: 10                             # 每分钟限制请求数 (可选)
     max_iterations: 30                  # 最大迭代次数
     required_verifications: 3           # 需要的成功验证次数
+    parallel_check: true                # 并行验证模式 (同时启动3个验证LLM调用)
     feature:
       vision: true                      # 视觉能力
       summary_think: true               # 生成思维链摘要

engine/deep_think.py

@@ -46,6 +46,7 @@ def __init__(
         model_stages: Dict[str, str] = None,
         on_progress: Optional[Callable[[ProgressEvent], None]] = None,
         enable_planning: bool = False,
+        enable_parallel_check: bool = False,
         llm_params: Optional[Dict[str, Any]] = None,
     ):
         self.client = client
@@ -62,6 +63,7 @@ def __init__(
         self.on_progress = on_progress
         self.sources: List[Source] = []
         self.enable_planning = enable_planning
+        self.enable_parallel_check = enable_parallel_check
         self.llm_params = llm_params or {}
         self._task = None  # 用于存储当前任务，以便取消
 
@@ -158,6 +160,93 @@ async def _verify_solution(
 
         return {"bug_report": bug_report, "good_verify": good_verify}
 
+    async def _verify_solution_parallel(
+        self,
+        problem_statement: MessageContent,
+        solution: str
+    ) -> Dict[str, str]:
+        """并行验证解决方案 - 同时启动required_verifications个验证LLM调用，全部通过才算成功"""
+        detailed_solution = self._extract_detailed_solution(solution)
+        # 提取文本用于构建提示词
+        problem_text = extract_text_from_content(problem_statement)
+        verification_prompt = build_verification_prompt(
+            problem_text,
+            detailed_solution
+        )
+        
+        num_checks = self.required_verifications
+        self._emit("progress", {"message": f"Parallel verifying solution ({num_checks} concurrent checks)..."})
+        
+        # 使用验证阶段的模型
+        verification_model = self._get_model_for_stage("verification")
+        
+        # 同时启动required_verifications个验证LLM调用
+        verification_tasks = [
+            self.client.generate_text(
+                model=verification_model,
+                system=VERIFICATION_SYSTEM_PROMPT,
+                prompt=verification_prompt,
+                **self.llm_params
+            )
+            for _ in range(num_checks)
+        ]
+        
+        # 等待所有验证完成
+        verification_outputs = await asyncio.gather(*verification_tasks)
+        
+        # 检查每个验证结果
+        check_tasks = []
+        for verification_output in verification_outputs:
+            check_prompt = (
+                f'Response in "yes" or "no". Is the following statement saying the '
+                f'solution is correct, or does not contain critical error or a major '
+                f'justification gap?\n\n{verification_output}'
+            )
+            check_tasks.append(
+                self.client.generate_text(
+                    model=verification_model,
+                    prompt=check_prompt,
+                    **self.llm_params
+                )
+            )
+        
+        # 等待所有检查完成
+        good_verifies = await asyncio.gather(*check_tasks)
+        
+        # 统计通过的验证数量
+        passed_count = sum(1 for gv in good_verifies if "yes" in gv.lower())
+        
+        # 需要全部通过才算验证成功
+        passed = passed_count == num_checks
+        
+        # 收集 bug 报告（如果有的话）
+        bug_reports = []
+        if not passed:
+            for i, (verification_output, good_verify) in enumerate(zip(verification_outputs, good_verifies)):
+                if "yes" not in good_verify.lower():
+                    bug_report = self._extract_detailed_solution(
+                        verification_output,
+                        "Detailed Review",
+                        False
+                    )
+                    if bug_report:
+                        bug_reports.append(f"[Check {i+1}] {bug_report}")
+        
+        combined_bug_report = "\n\n".join(bug_reports) if bug_reports else ""
+        
+        # 返回综合结果
+        good_verify_summary = f"yes (passed {passed_count}/{num_checks} checks)" if passed else f"no (passed {passed_count}/{num_checks} checks)"
+        
+        return {
+            "bug_report": combined_bug_report,
+            "good_verify": good_verify_summary,
+            "parallel_results": {
+                "total_checks": num_checks,
+                "passed_checks": passed_count,
+                "individual_results": good_verifies
+            }
+        }
+    
     async def _initial_exploration(
         self,
         problem_statement: MessageContent,
@@ -242,11 +331,17 @@ async def _initial_exploration(
 
         self._emit("solution", {"solution": improved_solution, "iteration": 0})
 
-        # 验证
-        verification = await self._verify_solution(
-            problem_statement,
-            improved_solution
-        )
+        # 验证 - 根据配置选择串行或并行
+        if self.enable_parallel_check:
+            verification = await self._verify_solution_parallel(
+                problem_statement,
+                improved_solution
+            )
+        else:
+            verification = await self._verify_solution(
+                problem_statement,
+                improved_solution
+            )
 
         self._emit("verification", {
             "passed": "yes" in verification["good_verify"].lower(),
@@ -400,8 +495,11 @@ async def run(self) -> DeepThinkResult:
                         knowledge_enhanced=len(self.sources) > 0,
                     )
 
-                # 再次验证
-                verification = await self._verify_solution(self.problem_statement, solution)
+                # 再次验证 - 根据配置选择串行或并行
+                if self.enable_parallel_check:
+                    verification = await self._verify_solution_parallel(self.problem_statement, solution)
+                else:
+                    verification = await self._verify_solution(self.problem_statement, solution)
                 self._emit("verification", {
                     "passed": "yes" in verification["good_verify"].lower(),
                     "iteration": i + 1,

engine/ultra_think.py

@@ -43,6 +43,7 @@ def __init__(
         model_stages: Dict[str, str] = None,
         on_progress: Optional[Callable[[ProgressEvent], None]] = None,
         on_agent_update: Optional[Callable[[str, Dict[str, Any]], None]] = None,
+        enable_parallel_check: bool = False,
         llm_params: Optional[Dict[str, Any]] = None,
     ):
         self.client = client
@@ -60,6 +61,7 @@ def __init__(
         self.model_stages = model_stages or {}
         self.on_progress = on_progress
         self.on_agent_update = on_agent_update
+        self.enable_parallel_check = enable_parallel_check
         self.sources: List[Source] = []
         self.llm_params = llm_params or {}
 
@@ -206,6 +208,7 @@ def agent_progress_handler(event: ProgressEvent):
                 max_errors_before_give_up=self.max_errors,
                 model_stages=self.model_stages,
                 on_progress=agent_progress_handler,
+                enable_parallel_check=self.enable_parallel_check,
                 llm_params=self.llm_params,
             )
 
@@ -340,6 +343,7 @@ def synthesis_progress_handler(event: ProgressEvent):
                 max_errors_before_give_up=self.max_errors,
                 model_stages=self.model_stages,
                 on_progress=synthesis_progress_handler,
+                enable_parallel_check=self.enable_parallel_check,
                 llm_params=self.llm_params,
             )