LangGraph范式-web

Web Voyager是一个基于多模态大模型（Multi-modal Agent）的AI Agent，旨在通过浏览器API与网页交互，完成用户提出的任务。其核心功能是根据用户的“问题、图像或请求”（Question, Image, or Request），生成网页操作的“工具响应”（Tool Responses），并最终返回“最终答案”（Final Answer）。

它的工作原理是查看每次轮次带注释的浏览器屏幕截图，然后选择下一步操作。该代理架构是一个基本的推理和行动 (ReAct) 循环。该代理的独特之处在于：

它使用类似标记集的图像注释作为代理的 UI 功能。

它通过使用工具控制鼠标和键盘，在浏览器中实现应用。

整体设计如下所示：

从示意图来看，其流程包括：

• 输入：用户提供问题、图像或请求。
• 处理：
  • 多模态Agent（Multi-modal Agent）接收输入，分析网页内容（Webpage Object）。
  • Agent通过工具模式（Tool Modes）生成下一步操作（Type、Scroll、Click、Wait、Jump to Search、GoBack）。
  • 借助浏览器API执行操作，获取新的网页状态。
  • 输出：最终答案。

    

    定义状态（State）

    LangGraph的工作流需要定义一个状态对象，用于在节点之间传递信息。Web Voyager的状态可能包括：

    • 当前网页的截图（base64编码）。
    • 用户的请求。
    • 当前的操作历史（Trajectory）。
    • 工具调用的结果。

      from typing import TypedDict, List
      class WebVoyagerState(TypedDict):
          user_request: str  # 用户的请求
          webpage_screenshot: str  # 当前网页截图（base64编码）
          trajectory: List[str]  # 操作历史
          tool_response: str  # 工具调用的响应
          final_answer: str  # 最终答案

      (3) 定义工具（Tools）

      Web Voyager的核心是工具模式（Tool Modes），包括 Type、Scroll、Click 等操作。这些操作通过浏览器API（如Playwright）实现。

      @tool
      async def click_element(element_id: str) -> str:
          """点击网页中的某个元素"""
          async with async_playwright() as p:
              browser = await p.chromium.launch()
              page = await browser.new_page()
              # 假设已经在某个页面上
              await page.click(f"#{element_id}")
              screenshot = await page.screenshot()
              await browser.close()
              return base64.b64encode(screenshot).decode("utf-8")
      @tool
      async def type_text(text: str, element_id: str) -> str:
          """在网页的某个元素中输入文本"""
          async with async_playwright() as p:
              browser = await p.chromium.launch()
              page = await browser.new_page()
              await page.fill(f"#{element_id}", text)
              screenshot = await page.screenshot()
              await browser.close()
              return base64.b64encode(screenshot).decode("utf-8")
      @tool
      async def scroll_page(direction: str) -> str:
          """滚动网页"""
          async with async_playwright() as p:
              browser = await p.chromium.launch()
              page = await browser.new_page()
              if direction == "down":
                  await page.evaluate("window.scrollBy(0, window.innerHeight)")
              screenshot = await page.screenshot()
              await browser.close()
              return base64.b64encode(screenshot).decode("utf-8")
      # 其他工具：wait、jump_to_search、go_back 等

      (4) 定义多模态Agent

      多模态Agent基于用户请求和网页截图，决定下一步操作。它通常使用一个提示模板（PromptTemplate）来引导模型推理。

      定义提示模板
      prompt = PromptTemplate(
          input_variables=["user_request", "webpage_screenshot", "trajectory"],
          template="""
          你是一个网页导航Agent。用户请求：{user_request}
          当前网页截图（base64编码）：{webpage_screenshot}
          操作历史：{trajectory}
          
          基于当前网页状态，决定下一步操作。可能的工具包括：
          - click_element(element_id): 点击某个元素
          - type_text(text, element_id): 输入文本
          - scroll_page(direction): 滚动页面
          - wait(): 等待
          - jump_to_search(): 跳转到搜索
          - go_back(): 返回上一页
          
          返回下一步操作的工具调用，例如：{{"tool": "click_element", "args": {{"element_id": "button1"}}}}
          如果任务已完成，返回最终答案，例如：{{"final_answer": "任务完成"}}
          """
      )
      # 初始化多模态模型
      llm = ChatOpenAI(model="gpt-4o", temperature=0)  # 使用支持多模态的模型
      agent = llm.bind_tools([click_element, type_text, scroll_page])

      (5) 定义LangGraph工作流

      LangGraph的工作流由节点和边组成。Web Voyager可能包含以下节点：

      • agent_node：调用多模态Agent，生成下一步操作。
      • tool_node：执行工具操作，更新网页状态。
      • end_node：输出最终答案。

        定义图
        workflow = StateGraph(WebVoyagerState)
        # 节点：Agent决策
        def agent_node(state: WebVoyagerState) -> WebVoyagerState:
            response = agent.invoke({
                "user_request": state["user_request"],
                "webpage_screenshot": state["webpage_screenshot"],
                "trajectory": state["trajectory"]
            })
            if "final_answer" in response:
                state["final_answer"] = response["final_answer"]
                return state
            state["tool_response"] = response
            return state
        # 节点：工具执行
        async def tool_node(state: WebVoyagerState) -> WebVoyagerState:
            tool_call = state["tool_response"]
            tool_name = tool_call["tool"]
            args = tool_call["args"]
            
            if tool_name == "click_element":
                screenshot = await click_element(args["element_id"])
            elif tool_name == "type_text":
                screenshot = await type_text(args["text"], args["element_id"])
            elif tool_name == "scroll_page":
                screenshot = await scroll_page(args["direction"])
            # 其他工具调用...
            
            state["webpage_screenshot"] = screenshot
            state["trajectory"].append(f"执行工具：{tool_name}，参数：{args}")
            return state
        # 添加节点
        workflow.add_node("agent", agent_node)
        workflow.add_node("tool", tool_node)
        # 添加边
        workflow.set_entry_point("agent")
        workflow.add_conditional_edges(
            "agent",
            lambda state: "end" if state.get("final_answer") else "tool",
            {"end": END, "tool": "tool"}
        )
        workflow.add_edge("tool", "agent")
        # 编译图
        app = workflow.compile()

        Paragoger衍生者AI训练营