app.py

import gradio as gr
import time
import base64
from datetime import datetime
import argparse
import os

from agent import BrowserAgent
from computers import BrowserbaseComputer, PlaywrightComputer
import base64
import sys

sys.path.append(os.path.dirname(__file__))

from prompt import prompt_options

PLAYWRIGHT_SCREEN_SIZE = (1440, 900)

os.environ["PLAYWRIGHT_HEADLESS"] = "false"

def run_genflow(query:str, prompt:str = ""):
    env = PlaywrightComputer(
        screen_size=PLAYWRIGHT_SCREEN_SIZE,
        initial_url="https://wenku.baidu.com/ndcore/browse/aiunion?fr=options_AIcard_1&_wkts_=1761290807747&bdQuery=genflow&t=1761290807744&tabType=genflow&aiCreat=genflow",
        # initial_url="https://www.doubao.com/chat/",
        highlight_mouse=True, # 如果指定，代理将尝试在屏幕截图中突出显示鼠标光标的位置。这对于可视化调试很有用。
    )
    # query="在提示文本为'输入问题，交给GenFlow的'搜索条中输入‘百度’，点击发送按钮",
    
    prompt = "GenFlow是一个AI聊天机器人。你需要作为测试员，在它的网页底部搜索栏中输入用户提交的问题, 并且观察其输出结果。" + prompt
    prompt += """
GenFlow 返回的内容可能比较长，你可以多次执行'scroll_at'操作来查看网页中它返回的上下文。
GenFlow 有时会呈现左边主页面，右边预览区域。在这种情况下执行'scroll_at'的时候你要注意鼠标位置。
通过反复的滚动，确保主页面和预览区域已经滚动到底。预览区域往往很长，所以你需要多次执行'scroll_at'操作，确保2次滚动操作看到的网页完全一样为止, 给出最终客观评价.

GenFlow 一定会给出'输出结果'，请保持足够的耐心！
**无论输入的语言是什么，你需要输出中文**"""
    # print(prompt)
    with env as browser_computer:
        agent = BrowserAgent(
            browser_computer=browser_computer,
            query=query,
            system_prompt=prompt,
            model_name='gemini-2.5-computer-use-preview-10-2025',
        )
        # reasoning, status, function_responses_list
        for step in agent.agent_loop_yield():
            # print(step)
            # input("☕️")
            yield step

# === 处理用户查询 ===
def process_user_query(query, prompt=""):
    if not query.strip():
        yield [], "请输入有效的查询", "<p>无执行步骤</p>"
        return

    steps = []

    for reasoning, status, function_responses_list in run_genflow(query, prompt):
        # 每个 function_responses_list 是若干个 ["screenshot", "action", "response"]
        sub_steps = []
        for item in function_responses_list:
            if len(item) == 3:
                screenshot_base64, action, response = item["screenshot"], item["action"], item["response"]
            else:
                screenshot_base64, action, response = "", "未知操作", ""

            sub_steps.append({
                "screenshot": f"data:image/png;base64,{base64.b64encode(screenshot_base64).decode("utf-8")}" if screenshot_base64 else "",
                "action": action,
                "response": response,
            })

        step = {
            "step": len(steps) + 1,
            "reasoning": reasoning,
            "status": status,
            "functions": sub_steps,  # ✅ 支持多个函数结果
        }
        steps.append(step)
        display_html = update_steps_display(steps)
        yield steps, f"正在执行第 {len(steps)} 步: {status}", display_html
        time.sleep(0.3)

    yield steps, f"任务完成！共执行 {len(steps)} 步。", update_steps_display(steps)

# === 更新步骤展示的 HTML ===
def update_steps_display(steps):
    """以HTML格式显示每一步的推理、函数动作及截图"""
    if not steps:
        return "<p>暂无执行步骤</p>"

    html = "<div style='font-family: Arial, sans-serif;'>"
    for step in steps:
        status_color = {
            "COMPLETE": "green",
            "CONTINUE": "orange",
            "FAILURE": "red",
        }.get(step["status"], "black")

        html += f"""
        <div style='border:1px solid #ddd; margin:10px 0; padding:15px; border-radius:8px;'>
            <div style='display:flex; justify-content:space-between; align-items:center;'>
                <h3 style='margin:0;'>步骤 {step['step']}</h3>
                <span style='color:{status_color}; font-weight:bold;'>{step['status']}</span>
            </div>
            <p><strong>推理:</strong> {step['reasoning']}</p>
        """

        # ✅ 支持多个函数动作展示
        for idx, func in enumerate(step["functions"], start=1):
            # print(func)
            html += f"""
            <div style="word-wrap: break-word; word-break: break-all; white-space: pre-wrap; border:1px solid #ddd; border-radius:6px; padding:8px; margin-top:6px;">
                <p><strong>函数调用 {idx}:</strong></p>
                <p>动作: {func['action']}</p>
                <p>返回: {func['response']}</p>
            """
            if func["screenshot"]:
                html += f"<img src='{func['screenshot']}' style='max-width:100%; border:1px solid #ccc; border-radius:6px;'/>"
            html += "</div>"

        html += "</div>"  # 结束step块
    html += "</div>"
    return html



def resolve_prompt(selected_value):
    """如果选择的是预定义 key，则返回对应 value，否则直接返回用户输入"""
    if selected_value in prompt_options:
        return prompt_options[selected_value]
    return selected_value or ""

# === 创建 Gradio 界面 ===
def create_demo_interface():
    with gr.Blocks(title="CUA 在线试用系统", theme=gr.themes.Soft()) as demo:
        gr.Markdown("""
<div style="font-family: 'Microsoft YaHei', sans-serif; font-size: 18px; line-height: 1.6;">
    <h1 style="color: #2E86AB; font-size: 28px; font-weight: bold; text-align: center;">🧭 CUA 在线试用系统</h1>
    <p style="color: #555; font-size: 16px; text-align: center;">
        输入一个任务描述，Agent 将自动进行 genflow 中浏览、截图、分析并执行下一步操作。
    </p>
</div>
""")

        with gr.Row():
            with gr.Column(scale=1):
                query_input = gr.Textbox(
                    label="需要在 genflow 中测试的问题",
                    placeholder="例如：帮我搜索AI新闻 / 购买一台笔记本电脑",
                    lines=2)
                
                prompt_dropdown = gr.Dropdown(
                    label="选择或在右侧系统提示词中输入评估标准",
                    choices=list(prompt_options.keys()),
                    value=None,
                    allow_custom_value=False,  # ✅ 允许用户手动输入文本
                )
                submit_btn = gr.Button("开始执行", variant="primary")
                
            with gr.Column(scale=2):
                status_output = gr.Textbox(label="执行状态", interactive=False)
                final_prompt = gr.Textbox(label="系统提示词", interactive=True,lines=5)
                # 按钮或自动触发逻辑
                prompt_dropdown.change(resolve_prompt, inputs=prompt_dropdown, outputs=final_prompt)


            

        # ✅ 把详细执行过程放在 JSON 前面
        with gr.Accordion("详细执行过程", open=True):
            steps_display = gr.HTML()

        # 再放执行步骤 JSON
        steps_output = gr.JSON(label="执行步骤详情（JSON）")

        # 流式输出绑定
        submit_btn.click(
            fn=process_user_query,
            inputs=[query_input, final_prompt],
            outputs=[steps_output, status_output, steps_display],
            api_name="run_agent",
            show_progress=True,
            queue=True,  # 支持流式输出
        )

    return demo



if __name__ == "__main__":
    demo = create_demo_interface()
    demo.queue()  # 必须启用 queue 才能支持 yield
    demo.launch(
        server_name="0.0.0.0",
        server_port=7860,
        share=True,
    )