LangChainとLlamaIndex: Webページローダーとread_urlツール

import os

import requests

API = "https://webreaderapi.com/api/v1/read"
HEADERS = {"Authorization": "Bearer " + os.environ["WEBREADER_API_KEY"]}


class PageError(Exception):
    pass


def read_page(url, **options):
    resp = requests.get(API, headers=HEADERS, params={"url": url, **options}, timeout=75)
    body = resp.json()
    if resp.status_code != 200:
        raise PageError(body["error"]["code"])
    return body

from langchain_core.documents import Document
from langchain_text_splitters import MarkdownHeaderTextSplitter

from reader import PageError, read_page

splitter = MarkdownHeaderTextSplitter(headers_to_split_on=[("#", "h1"), ("##", "h2"), ("###", "h3")])


def load_pages(urls):
    chunks = []
    for url in urls:
        try:
            page = read_page(url)
        except PageError as exc:
            print("skipped", url, exc)
            continue
        for part in splitter.split_text(page["markdown"]):
            part.metadata.update({"source": page["final_url"], "title": page.get("title", "")})
            chunks.append(part)
    return chunks


docs = load_pages(["https://example.com/docs/install", "https://example.com/docs/configure"])
print(len(docs), "chunks", isinstance(docs[0], Document) if docs else "")

import os

from langchain.chat_models import init_chat_model
from langchain_core.tools import tool

from reader import PageError, read_page


@tool
def read_url(url: str, max_chars: int = 20000) -> str:
    """Read a public web page and return its main content as Markdown, with its title and source URL."""
    try:
        page = read_page(url, max_chars=max_chars)
    except PageError as exc:
        return "Could not read the page: " + str(exc)
    return "# " + (page.get("title") or url) + "\nSource: " + page["final_url"] + "\n\n" + page["markdown"]


model = init_chat_model(os.environ.get("CHAT_MODEL", "openai:gpt-4.1-mini")).bind_tools([read_url])
reply = model.invoke("Summarise https://example.com/blog/first-post in three bullet points.")
print(reply.tool_calls)

import os
import time

import requests
from llama_index.core import Document, VectorStoreIndex
from llama_index.core.tools import FunctionTool

from reader import HEADERS, PageError, read_page


def crawl_pages(start_url, limit=200):
    job = requests.post("https://webreaderapi.com/api/v1/crawl", headers=HEADERS, json={"url": start_url, "limit": limit}, timeout=30).json()
    pages, cursor = [], None
    while True:
        params = {"cursor": cursor} if cursor else {}
        res = requests.get("https://webreaderapi.com/api/v1/jobs/" + job["job_id"], headers=HEADERS, params=params, timeout=30).json()
        pages += [r for r in res["results"] if "error" not in r]
        cursor = res.get("next_cursor")
        if not cursor and res["status"] == "done":
            return pages
        if not cursor:
            time.sleep(5)


pages = crawl_pages("https://docs.example.com/")
docs = [Document(text=p["markdown"], metadata={"source": p["final_url"], "title": p.get("title", "")}) for p in pages]
index = VectorStoreIndex.from_documents(docs)
print(index.as_query_engine().query("How do I install the product?"))


def read_url(url: str, max_chars: int = 20000) -> str:
    """Read a public web page and return its main content as Markdown."""
    try:
        return read_page(url, max_chars=max_chars)["markdown"]
    except PageError as exc:
        return "Could not read the page: " + str(exc)


read_tool = FunctionTool.from_defaults(fn=read_url, name="read_url")
print(os.environ.get("CHAT_MODEL", "set CHAT_MODEL for your agent"), read_tool.metadata.name)