Function run_self_reflection_async

src/webwright/tools/self_reflection.py:379–447

(
    *,
    images: list[Path],
    image_judge_system_prompt: str,
    image_judge_user_prompt: str,
    final_verdict_system_prompt: str,
    final_verdict_user_prompt: str,
    action_history_log: str,
    max_image_parse_retries: int,
    final_max_new_tokens: int,
    image_max_new_tokens: int,
    model_client: Any,
)

async def run_self_reflection_async(
    *,
    images: list[Path],
    image_judge_system_prompt: str,
    image_judge_user_prompt: str,
    final_verdict_system_prompt: str,
    final_verdict_user_prompt: str,
    action_history_log: str,
    max_image_parse_retries: int,
    final_max_new_tokens: int,
    image_max_new_tokens: int,
    model_client: Any,
) -> SelfReflectionResult:
    model_name = str(getattr(model_client.config, "model_name", ""))
    endpoint = _model_endpoint(model_client)

    if images:
        per_image = await asyncio.gather(
            *(
                _judge_one_image(
                    image_path=path,
                    image_judge_system_prompt=image_judge_system_prompt,
                    image_judge_user_prompt=image_judge_user_prompt,
                    model_client=model_client,
                    max_new_tokens=image_max_new_tokens,
                    max_parse_retries=max_image_parse_retries,
                )
                for path in images
            )
        )
    else:
        per_image = []

    image_paths = [record["image_path"] for record in per_image]
    reasonings = [record["Reasoning"] or "" for record in per_image]

    reasonings_block = "\n".join(
        f"{i + 1}. {text}" for i, text in enumerate(reasonings)
    )

    final_user_text = _render_final_verdict_user_prompt(
        final_verdict_user_prompt,
        image_reasonings=reasonings_block,
        action_history_log=action_history_log,
    )

    user_content: list[dict[str, Any]] = [text_part(final_user_text)]
    for path_str in image_paths:
        user_content.append(_high_detail_image_part_from_path(Path(path_str)))

    final_response = await asyncio.to_thread(
        _call_model,
        model_client=model_client,
        system_prompt=final_verdict_system_prompt,
        user_content=user_content,
        max_new_tokens=final_max_new_tokens,
    )
    predicted_label = _parse_final_verdict(final_response)