Olostep + LangChain 統合

pip install langchain-olostep

poetry add langchain-olostep

from langchain_olostep import scrape_website
import asyncio

# ウェブサイトをスクレイプ
content = asyncio.run(scrape_website.ainvoke({
    "url": "https://example.com",
    "format": "markdown"
}))

print(content)

# 動的コンテンツを待機
content = asyncio.run(scrape_website.ainvoke({
    "url": "https://example.com",
    "format": "markdown",
    "wait_before_scraping": 2000
}))

# 特殊なパーサーを使用
content = asyncio.run(scrape_website.ainvoke({
    "url": "https://www.amazon.com/dp/PRODUCT_ID",
    "parser": "@olostep/amazon-product",
    "format": "json"
}))

from langchain_olostep import scrape_batch
import asyncio

# 複数のURLをスクレイプ
result = asyncio.run(scrape_batch.ainvoke({
    "urls": [
        "https://example1.com",
        "https://example2.com",
        "https://example3.com"
    ],
    "format": "markdown"
}))

print(result)
# 戻り値: {"batch_id": "batch_xxx", "status": "in_progress", ...}

from langchain_olostep import answer_question
import asyncio

# 簡単な質問をする
result = asyncio.run(answer_question.ainvoke({
    "task": "What is the capital of France?"
}))

print(result)
# 戻り値: {"answer": {"result": "Paris"}, "sources": [...]}

# JSONスキーマで構造化データを取得
result = asyncio.run(answer_question.ainvoke({
    "task": "What is the latest book by J.K. Rowling?",
    "json_schema": {
        "book_title": "",
        "author": "",
        "release_date": ""
    }
}))

print(result)
# ソース付きの構造化された回答を返す

# 会社データを強化
result = asyncio.run(answer_question.ainvoke({
    "task": "Find the CEO and headquarters of Stripe",
    "json_schema": {
        "ceo_name": "",
        "headquarters": "",
        "founded_year": ""
    }
}))

# 不確実性を"NOT_FOUND"値で処理

from langchain_olostep import extract_urls
import asyncio

# ウェブサイトからすべてのURLを取得
result = asyncio.run(extract_urls.ainvoke({
    "url": "https://example.com",
    "top_n": 100
}))

print(result)
# 戻り値: {"urls": [...], "total_urls": 100, ...}

# ブログURLのみを取得
result = asyncio.run(extract_urls.ainvoke({
    "url": "https://example.com",
    "include_urls": ["/blog/**"],
    "exclude_urls": ["/admin/**", "/private/**"],
    "top_n": 50
}))

from langchain_olostep import crawl_website
import asyncio

# ドキュメントサイト全体をクロール
result = asyncio.run(crawl_website.ainvoke({
    "start_url": "https://docs.example.com",
    "max_pages": 100
}))

print(result)
# 戻り値: {"crawl_id": "crawl_xxx", "status": "in_progress", ...}

# URLフィルタを使用してクロール
result = asyncio.run(crawl_website.ainvoke({
    "start_url": "https://example.com",
    "max_pages": 200,
    "include_urls": ["/**"],
    "exclude_urls": ["/admin/**", "/private/**"],
    "max_depth": 3
}))

from langchain.agents import initialize_agent, AgentType
from langchain_openai import ChatOpenAI
from langchain_olostep import (
    scrape_website,
    answer_question,
    extract_urls
)

# Olostepツールでエージェントを作成
tools = [scrape_website, answer_question, extract_urls]
llm = ChatOpenAI(model="gpt-4o-mini")

agent = initialize_agent(
    tools=tools,
    llm=llm,
    agent=AgentType.ZERO_SHOT_REACT_DESCRIPTION,
    verbose=True
)

# エージェントを使用
result = agent.run("""
Research the company at https://company.com:
1. Scrape their about page
2. Search for their latest funding round
3. Extract all their product pages
""")

print(result)

from langgraph.graph import StateGraph, END
from langchain_olostep import (
    scrape_website,
    scrape_batch,
    answer_question,
    extract_urls
)
from langchain_openai import ChatOpenAI
import json

def create_research_agent():
    workflow = StateGraph(dict)
    
    def discover_pages(state):
        # ターゲットサイトからすべてのURLを抽出
        result = extract_urls.invoke({
            "url": state["target_url"],
            "include_urls": ["/product/**"],
            "top_n": 50
        })
        state["urls"] = json.loads(result)["urls"]
        return state
    
    def scrape_pages(state):
        # 発見されたページをバッチでスクレイプ
        result = scrape_batch.invoke({
            "urls": state["urls"],
            "format": "markdown"
        })
        state["batch_id"] = json.loads(result)["batch_id"]
        return state
    
    def answer_questions(state):
        # データに関する質問にAIを使用して回答
        result = answer_question.invoke({
            "task": state["research_question"],
            "json_schema": state["desired_format"]
        })
        state["answer"] = json.loads(result)["answer"]
        return state
    
    workflow.add_node("discover", discover_pages)
    workflow.add_node("scrape", scrape_pages)
    workflow.add_node("analyze", answer_questions)
    
    workflow.set_entry_point("discover")
    workflow.add_edge("discover", "scrape")
    workflow.add_edge("scrape", "analyze")
    workflow.add_edge("analyze", END)
    
    return workflow.compile()

# エージェントを使用
agent = create_research_agent()
result = agent.invoke({
    "target_url": "https://store.com",
    "research_question": "What are the top 5 most expensive products?",
    "desired_format": {
        "products": [{"name": "", "price": "", "url": ""}]
    }
})