使用Lybic的Gradio Playground示例

mkdir gradio-playground
cd gradio-playground
python -m venv venv
source venv/bin/activate  # 在Linux或macOS上
venv\Scripts\activate  # 在Windows上
pip install -U pip
pip install -U python-dotenv langchain-core langchain-google-genai langchain-openai langchain gradio lybic

GOOGLE_API_KEY=your_google_api_key_here

ARK_API_KEY=your_api_key
ARK_MODEL_ENDPOINT=https://ark.cn-beijing.volces.com/api/v3
ARK_MODEL_NAME=doubao-1-5-ui-tars-250428

import os
import dotenv

dotenv.load_dotenv(os.path.join(os.path.dirname(__file__), ".env"))

from langchain_google_genai import ChatGoogleGenerativeAI
from langchain_core.prompts import PromptTemplate
from langchain.memory import ConversationBufferMemory
from langchain.chains import LLMChain

import gradio as gr

llm = ChatGoogleGenerativeAI(model="gemini-2.5-flash", temperature=0.7)

prompt = PromptTemplate(
    input_variables=["history", "user_input"],
    template="You are a helpful assistant. Below is the conversation history:\n{history}\n\nUser input: {user_input}\n\nAssistant response:"
)

memory = ConversationBufferMemory(memory_key="history")
chain = LLMChain(llm=llm, prompt=prompt, memory=memory)

def playground(user_input, history=None):
    if not user_input or user_input.strip() == "":
        return "It looks like you didn't provide any text for me to respond to! Please type or paste your input, and I'll be happy to help.", history or []
    try:
        response = chain.invoke({"user_input": user_input})["text"]
        new_history = history + [[user_input, response]] if history else [[user_input, response]]
        return "", new_history
    except Exception as e:
        return f"Error: {str(e)}", history

with gr.Blocks() as iface:
    gr.Markdown("# Playground with LangChain & Gradio")
    gr.Markdown("Interact with the Gemini model and maintain conversation history.")

    chatbot = gr.Chatbot()
    user_input = gr.Textbox(placeholder="Enter your question or prompt here...")
    submit_button = gr.Button("Submit")

    submit_button.click(
        fn=playground,
        inputs=[user_input, chatbot],
        outputs=[user_input, chatbot],
    )

    user_input.submit(
        fn=playground,
        inputs=[user_input, chatbot],
        outputs=[user_input, chatbot],
    )

if __name__ == "__main__":
    iface.launch()

python app.py

from lybic import LybicClient
async def main():
  # you should set env vars before initialize LybicClient
  async with LybicClient() as client:
    pass

import asyncio
from lybic import LybicClient, Sandbox

async def main():
  async with LybicClient() as client:
      sandbox = Sandbox(client)
      url, image, b64_str = await sandbox.get_screenshot(sandbox_id="SBX-xxxx")
      print(f"Screenshot URL: {url}")
      image.show()

if __name__ == '__main__':
      asyncio.run(main())

from langchain_core.messages import SystemMessage, AIMessage, HumanMessage
messages = [
    HumanMessage(
    content=[
        {
            "type": "text",
            "text": "What is this? Please describe the picture in English."
        },
        {
            "type": "image_url",
            "image_url": {
                "url": f"data:image/webp;base64,{ b64_str }"
            }
        }
    ]
),
]

You are a GUI Planner agent. You need to perform the next action to complete the task.

You have a access to a sandbox environment which are running Windows OS. You should ensure your action follows Windows common sense, such as double click to open a software on desktop, or use right click for context menu, and so on.

You should describe the next ONE action in English, such as:

- double click at "Google Chrome" icon
- right click at "This PC" icon
- click the close button at the top right corner
- press Alt+F4
- type "terms"
- wait 5 seconds

After that, an agent will execute the action you describe, and a new screenshot will be attached, you can use it to plan the next step.

You are a GUI Grounding Agent, proficient in the operation of various commonly used software on Windows, Linux, and other operating systems.
Please complete the GUI Planner Agent's task based on its input, history Action, and screenshots.
The user(GUI Planner Agent) will input a current screenshot and a text including `ActionSummary` and `Action`.
You need to analyze and process `ActionSummary` and `Action` from GUI Planner Agent, and output only one Action at a time, please strictly follow the format below.

## Output Format

Thought: ...
Action: ...

## Action Space
click(point='<point>x1 y1</point>')
left_double(point='<point>x1 y1</point>')
right_single(point='<point>x1 y1</point>')
drag(start_point='<point>x1 y1</point>', end_point='<point>x2 y2</point>')
hotkey(key='ctrl c') # Split keys with a space and use lowercase. Also, do not use more than 3 keys in one hotkey action.
type(content='xxx') # Use escape characters \', \", and \n in content part to ensure we can parse the content in normal python string format. If you want to submit your input, use \n at the end of content, and next action use hotkey(key='enter')
scroll(point='<point>x1 y1</point>', direction='down or up or right or left') # Show more information on the `direction` side.
wait() #Sleep for 5s and take a screenshot to check for any changes.
finished(content='xxx') # Use escape characters \', \", and \n in content part to ensure we can parse the content in normal python string format.

## Note
- Use English in `Thought` part.
- The x1,x2 and y1,y2 are the coordinates of the element.
- The resolution of the screenshot is 1280*720.
- Write a small plan and finally summarize your next action (with its target element) in one sentence in `Thought` part.

from langchain_openai import ChatOpenAI
llm_uitars = ChatOpenAI(
  base_url=os.getenv("ARK_MODEL_ENDPOINT"),
  api_key=os.getenv("ARK_API_KEY"),
  model=os.getenv("ARK_MODEL_NAME"),
)

import asyncio
import os
import dotenv
import re

dotenv.load_dotenv(os.path.join(os.path.dirname(__file__), ".env"))

from langchain_google_genai import ChatGoogleGenerativeAI
from langchain_core.messages import SystemMessage, HumanMessage
from langchain.memory import ConversationBufferMemory
from lybic import LybicClient, Sandbox, ComputerUse, dto

import gradio as gr

PLANNER_SYSTEM_PROMPT = """You are a GUI Planner agent. You need to perform the next action to complete the task.

You have a access to a sandbox environment which are running Windows OS. You should ensure your action follows Windows common sense, such as double click to open a software on desktop, or use right click for context menu, and so on.

You should describe the next ONE action in English, such as:

- double click at "Google Chrome" icon
- right click at "This PC" icon
- click the close button at the top right corner
- press Alt+F4
- type "terms"
- wait 5 seconds

After that, an agent will execute the action you describe, and a new screenshot will be attached, you can use it to plan the next step.
"""

GROUNDNING_SYSTEM_PROMPT = """You are a GUI Grounding Agent, proficient in the operation of various commonly used software on Windows, Linux, and other operating systems.
Please complete the GUI Planner Agent's task based on its input, and screenshots.
The user(GUI Planner Agent) will input a current screenshot and an actio text.
You need to analyze and process `action` from GUI Planner Agent, and output only one Action at a time, please strictly follow the format below.

## Output Format

Thought: ...
Action: ...

## Action Space
click(point='<point>x1 y1</point>')
left_double(point='<point>x1 y1</point>')
right_single(point='<point>x1 y1</point>')
drag(start_point='<point>x1 y1</point>', end_point='<point>x2 y2</point>')
hotkey(key='ctrl c') # Split keys with a space and use lowercase. Also, do not use more than 3 keys in one hotkey action.
type(content='xxx') # Use escape characters \', \", and \n in content part to ensure we can parse the content in normal python string format. If you want to submit your input, use \n at the end of content
scroll(point='<point>x1 y1</point>', direction='down or up or right or left') # Show more information on the `direction` side.
wait() #Sleep for 5s and take a screenshot to check for any changes.
finished(content='xxx') # Use escape characters \', \", and \n in content part to ensure we can parse the content in normal python string format.

## Note
- Use English in `Thought` part.
- The x1,x2 and y1,y2 are the coordinates of the element.
- The resolution of the screenshot is 1280*720.
- Write a small plan and finally summarize your next action (with its target element) in one sentence in `Thought` part.
"""

sandbox_id = os.getenv("SANDBOX")
from langchain_openai import ChatOpenAI
llm = ChatGoogleGenerativeAI(model="gemini-2.5-flash", temperature=0.7)
memory = ConversationBufferMemory(memory_key="history")
llm_uitars = ChatOpenAI(
  base_url=os.getenv("ARK_MODEL_ENDPOINT"),
  api_key=os.getenv("ARK_API_KEY"),
  model=os.getenv("ARK_MODEL_NAME"),
)

async def playground(user_input, history=None):
  if not user_input or user_input.strip() == "":
      yield (
          "It looks like you didn't provide any text for me to respond to! Please type or paste your input, and I'll be happy to help.",
          history or [],
      )
      return

  if history is None:
      history = []

  try:
      async with LybicClient() as client:
          sandbox = Sandbox(client)
          computer_use = ComputerUse(client)

          # Add the initial user input to the history
          screenshot_url, _, _ = await sandbox.get_screenshot(sandbox_id)
          user_message_for_history = f'{user_input}\n\n![screenshot]({screenshot_url})'
          history.append([user_message_for_history, "Thinking..."])
          yield "", history

          while True:
              # 1. Observe: Get the current screenshot
              screenshot_url, _, base64_str = await sandbox.get_screenshot(sandbox_id)

              # 2. Think: Prepare messages for the LLM
              planner_message = [
                  HumanMessage(
                      content=[
                          {"type": "text", "text": f"This is user's instruction: {user_input}"},
                          {
                              "type": "image_url",
                              "image_url": {"url": f"data:image/webp;base64,{base64_str}"},
                          },
                      ]
                  ),
              ]

              system_message = SystemMessage(content=PLANNER_SYSTEM_PROMPT)
              history_messages = memory.chat_memory.messages
              full_message_list = [system_message] + history_messages + planner_message

              # Get LLM response
              response_message = await llm.ainvoke(full_message_list)
              response = response_message.text()
              print("Planner output:", response)
              print("---------")

              # Update history with the LLM's thought process


              match = re.search(r'Action:\s*(.*)', response)
              if match:
                  response = match.group(1).strip()
              else:
                  yield f"Error: No action found", history
              history[-1][1] = response  # Update the "Thinking..." message
              yield "", history

              # 3. Grounding
              grounding_message = [
                  HumanMessage(
                      content=[
                          {"type": "text", "text": response},
                          {"type": "text",
                           "text": f"This is user's screenshot,and the resolution of the screenshot is 1280*720:"},
                          {
                              "type": "image_url",
                              "image_url": {"url": f"data:image/webp;base64,{base64_str}"},
                          }
                      ]
                  )
              ]

              system_message = SystemMessage(content=GROUNDNING_SYSTEM_PROMPT)
              full_message_list = [system_message] + grounding_message
              response_message = await llm_uitars.ainvoke(full_message_list)
              response = response_message.text()
              print("Grounding output:", response)

              # 4. Act: Parse and execute the action
              action_dto = await computer_use.parse_model_output(
                  dto.ComputerUseParseRequestDto(model="ui-tars",
                                                 textContent=response,
                                                 ))

              action = action_dto.actions[0]
              await computer_use.execute_computer_use_action(sandbox_id, dto.ComputerUseActionDto(action=action))

              # Memory: Store the conversation
              memory.chat_memory.add_message(planner_message[0])
              memory.chat_memory.add_ai_message(response)

              # Check for termination condition
              if isinstance(action, dto.FinishedAction):
                  final_screenshot_url, _, _ = await sandbox.get_screenshot(sandbox_id)
                  history.append([None, f"Task finished. Final state:\n\n![screenshot]({final_screenshot_url})"])
                  yield "", history
                  break

              await asyncio.sleep(2)
              # Prepare for the next loop iteration
              history.append([f"Action executed: `{ action.model_dump_json() }`\n\nTaking new screenshot...", "Thinking..."])
              yield "", history
  except Exception as e:
      import traceback
      yield f"Error: {str(e)}\n{traceback.format_exc()}", history

with gr.Blocks() as iface:
  gr.Markdown("# Playground with LangChain & Gradio")
  gr.Markdown("Interact with the Gemini model and maintain conversation history.")

  chatbot = gr.Chatbot(render_markdown=True, sanitize_html=False)
  user_input = gr.Textbox(placeholder="Enter your task for the agent...")
  submit_button = gr.Button("Run Agent")

  # Use a generator for the click event
  submit_button.click(
      fn=playground,
      inputs=[user_input, chatbot],
      outputs=[user_input, chatbot],
  )

  user_input.submit(
      fn=playground,
      inputs=[user_input, chatbot],
      outputs=[user_input, chatbot],
  )

if __name__ == "__main__":
  iface.launch()