Anchor Browser 集成

pip install --quiet -U langchain-anchorbrowser pydantic

import getpass
import os

if not os.environ.get("ANCHORBROWSER_API_KEY"):
    os.environ["ANCHORBROWSER_API_KEY"] = getpass.getpass("ANCHORBROWSER API key:\n")

from langchain_anchorbrowser import (
    AnchorContentTool,
    AnchorScreenshotTool,
    AdvancedAnchorWebTaskTool,
)

anchor_content_tool = AnchorContentTool()
anchor_screenshot_tool = AnchorScreenshotTool()
anchor_advanced_web_task_tool = AdvancedAnchorWebTaskTool()

# Get Markdown Content for https://www.anchorbrowser.io
anchor_content_tool.invoke(
    {"url": "https://www.anchorbrowser.io", "format": "markdown"}
)

# Get a Screenshot for https://docs.anchorbrowser.io
anchor_screenshot_tool.invoke(
    {"url": "https://docs.anchorbrowser.io", "width": 1280, "height": 720}
)

# Define a Pydantic model for the web task output schema

from pydantic import BaseModel
from typing import List

class NodeCpuUsage(BaseModel):
    node: str,
    cluster: str,
    cpu_avg_percentage: float

class OutputSchema(BaseModel):
    nodes_cpu_usage: List[NodeCpuUsage]

# Run a web task to collect data from a web page
anchor_advanced_web_task_tool.invoke(
    {
        "prompt": "Collect the node names and their CPU average %",
        "url": "https://play.grafana.org/a/grafana-k8s-app/navigation/nodes?from=now-1h&to=now&refresh=1m",
        "output_schema": OutputSchema.model_json_schema()
    }
)

# This is usually generated by a model, but we'll create a tool call directly for demo purposes.
model_generated_tool_call = {
    "args": {"url": "https://www.anchorbrowser.io", "format": "markdown"},
    "id": "1",
    "name": anchor_content_tool.name,
    "type": "tool_call",
}
anchor_content_tool.invoke(model_generated_tool_call)

pip install -qU langchain langchain-openai

from langchain.chat_models import init_chat_model

model = init_chat_model(model="gpt-4.1", model_provider="openai")

if not os.environ.get("OPENAI_API_KEY"):
    os.environ["OPENAI_API_KEY"] = getpass.getpass("OPENAI API key:\n")

from langchain_core.prompts import ChatPromptTemplate
from langchain_core.runnables import RunnableConfig, chain

prompt = ChatPromptTemplate(
    [
        ("system", "You are a helpful assistant."),
        ("human", "{user_input}"),
        ("placeholder", "{messages}"),
    ]
)

# specifying tool_choice will force the model to call this tool.
model_with_tools = model.bind_tools(
    [anchor_content_tool], tool_choice=anchor_content_tool.name
)

model_chain = prompt | model_with_tools


@chain
def tool_chain(user_input: str, config: RunnableConfig):
    input_ = {"user_input": user_input}
    ai_msg = model_chain.invoke(input_, config=config)
    tool_msgs = anchor_content_tool.batch(ai_msg.tool_calls, config=config)
    return model_chain.invoke({**input_, "messages": [ai_msg, *tool_msgs]}, config=config)


tool_chain.invoke(input())