LangChain1.0-Agent(进阶)本地模型+Playwright实现网页自动化操作

发布时间：2026-02-02 22:11:35编辑：123阅读（254）

LangChain 1.0 的核心价值是让大模型自主决策是否使用工具、使用哪种工具完成任务，Playwright 工具结合 Agent 后，可实现模型自主访问网页、提取所需信息、完成任务。

代码如下:

import asyncio
from playwright.async_api import async_playwright
from langchain_ollama import ChatOllama
from langchain.agents import create_agent
from langchain.tools import tool
from langchain_core.tools import ToolException


# ====================== 1. 浏览器状态  ======================
def create_browser_state():
    return {
        "playwright": None,
        "browser": None,
        "context": None,
        "page": None,
    }

# ====================== 2. 初始化浏览器 ======================
async def init_browser(state: dict):
    state["playwright"] = await async_playwright().start()
    state["browser"] = await state["playwright"].chromium.launch(headless=False)
    state["context"] = await state["browser"].new_context()
    state["page"] = await state["context"].new_page()
    print("✅ 异步浏览器初始化成功")


# ====================== 3. 关闭浏览器 ======================
async def close_browser(state: dict):
    if state.get("browser"):
        await state["browser"].close()
    if state.get("playwright"):
        await state["playwright"].stop()
    print("❌ 异步浏览器已关闭")


# ====================== 4. LangChain Tools（函数式） ======================
def create_browser_tools(state: dict):
    @tool
    async def open_url(url: str) -> str:
        """打开目标 URL"""
        page = state.get("page")
        if not page:
            raise ToolException("浏览器未初始化")

        try:
            await page.goto(url, wait_until="networkidle")
            title = await page.title()
            return f"成功打开 URL：{url}，页面标题：{title}"
        except Exception as e:
            raise ToolException(str(e))
    @tool
    async def click_element(selector: str) -> str:
        """点击页面元素（CSS / XPath）"""
        page = state.get("page")
        if not page:
            raise ToolException("浏览器未初始化")

        try:
            if selector.startswith("//"):
                await page.wait_for_selector(f"xpath={selector}", state="visible", timeout=10000)
                await page.click(f"xpath={selector}")
            else:
                await page.wait_for_selector(selector, state="visible", timeout=10000)
                await page.click(selector)
            return f"成功点击元素：{selector}"
        except Exception as e:
            raise ToolException(str(e))
    @tool
    async def input_text(selector: str, text: str) -> str:
        """输入文本"""
        page = state.get("page")
        if not page:
            raise ToolException("浏览器未初始化")

        try:
            if selector.startswith("//"):
                await page.fill(f"xpath={selector}", text)
            else:
                await page.fill(selector, text)
            return f"已输入文本：{text}"
        except Exception as e:
            raise ToolException(str(e))
    @tool
    async def extract_page_text() -> str:
        """提取页面文本"""
        page = state.get("page")
        if not page:
            raise ToolException("浏览器未初始化")

        try:
            text = await page.text_content("body")
            cleaned = "\n".join(
                line.strip() for line in text.split("\n") if line.strip()
            )
            return cleaned[:2000]
        except Exception as e:
            raise ToolException(str(e))

    return [
        open_url,
        click_element,
        input_text,
        extract_page_text,
    ]

# ====================== 5. 初始化 Agent ======================
def init_agent(tools):
    model = ChatOllama(
        model="llama3.1:8b",
        temperature=0,
    )
    system_prompt = """
你是一个网页自动化 AI Agent，只能使用提供的工具完成任务：
1. 先打开 URL
2. 再执行点击 / 输入
3. 必要时提取页面文本
4. 不允许编造操作
"""

    return create_agent(
        model=model,
        tools=tools,
        system_prompt=system_prompt,
        # debug=True,
    )

# ====================== 6. 主入口 ======================
async def main():
    state = create_browser_state()

    await init_browser(state)

    tools = create_browser_tools(state)
    agent = init_agent(tools)

    try:
        task = "打开 https://www.py3study.com 返回标题"
        print("🚀 开始执行任务")
        result = await agent.ainvoke(
            {"messages": [{"role": "user", "content": task}]}
        )
        print("\n🎯 执行结果：\n", result['messages'][-1].content)
    finally:
        await close_browser(state)


# ====================== 7. 程序入口 ======================
if __name__ == "__main__":
    asyncio.run(main())

运行结果：

✅ 异步浏览器初始化成功

🚀 开始执行任务




🎯 执行结果：

 页面标题为：Run博客 - 首页 - python3学习

❌ 异步浏览器已关闭

关键字：