> ## Documentation Index > Fetch the complete documentation index at: https://docs.olostep.com/llms.txt > Use this file to discover all available pages before exploring further. # Olostep + LangChain 集成 > 使用 LangChain 构建具有网页抓取和搜索功能的智能 AI 代理 Olostep LangChain 集成提供了全面的工具，帮助构建可以搜索、抓取、分析和结构化任何网站数据的 AI 代理。非常适合 LangChain 和 LangGraph 应用。 ## 功能该集成提供对所有 5 种 Olostep API 功能的访问：从任何单一 URL 提取内容，支持多种格式（Markdown、HTML、JSON、文本）并行处理多达 10,000 个 URL。批处理作业在 5-8 分钟内完成支持自然语言查询和结构化输出的 AI 驱动网页搜索提取网站的所有 URL 以进行站点结构分析通过跟随链接自主发现并抓取整个网站 ## 安装 ```bash pip theme={null} pip install langchain-olostep ``` ```bash poetry theme={null} poetry add langchain-olostep ``` ## 设置将你的 Olostep API 密钥设置为环境变量： ```bash theme={null} export OLOSTEP_API_KEY="your_olostep_api_key_here" ``` 从 [Olostep Dashboard](https://olostep.com/dashboard) 获取你的 API 密钥。 ## 可用工具 ### scrape\_website 从单一 URL 提取内容。支持多种格式和 JavaScript 渲染。要抓取的网站 URL（必须包括 http\:// 或 https\://）输出格式：`markdown`、`html`、`json` 或 `text` 用于特定位置内容的国家代码（例如，“US”、“GB”、“CA”） JavaScript 渲染的等待时间（0-10000 毫秒）用于专门提取的可选解析器 ID（例如，“@olostep/amazon-product”） ```python Basic Scraping theme={null} from langchain_olostep import scrape_website import asyncio # 抓取一个网站 content = asyncio.run(scrape_website.ainvoke({ "url": "https://example.com", "format": "markdown" })) print(content) ``` ```python With JavaScript theme={null} # 等待动态内容 content = asyncio.run(scrape_website.ainvoke({ "url": "https://example.com", "format": "markdown", "wait_before_scraping": 2000 })) ``` ```python With Parser theme={null} # 使用专门的解析器 content = asyncio.run(scrape_website.ainvoke({ "url": "https://www.amazon.com/dp/PRODUCT_ID", "parser": "@olostep/amazon-product", "format": "json" })) ``` ### scrape\_batch 并行处理多个 URL（一次最多 10,000 个）。要抓取的 URL 列表所有 URL 的输出格式：`markdown`、`html`、`json` 或 `text` 用于特定位置内容的国家代码 JavaScript 渲染的等待时间用于专门提取的可选解析器 ID ```python Batch Scraping theme={null} 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", ...} ``` ### answer\_question 搜索网络并获取带有来源的 AI 驱动答案。非常适合数据丰富和研究。要搜索的问题或任务描述所需输出格式的可选 JSON 架构字典/字符串 ```python Simple Question theme={null} 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": [...]} ``` ```python Structured Output theme={null} # 使用 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) # 返回带有来源的结构化答案 ``` ```python Data Enrichment theme={null} # 丰富公司数据 result = asyncio.run(answer_question.ainvoke({ "task": "Find the CEO and headquarters of Stripe", "json_schema": { "ceo_name": "", "headquarters": "", "founded_year": "" } })) # 使用 "NOT_FOUND" 值处理不确定性 ``` ### extract\_urls 提取网站的所有 URL 以进行站点结构分析。要从中提取 URL 的网站 URL 用于过滤 URL 的可选搜索查询限制返回的 URL 数量要包含的全局模式（例如，\["/blog/\*\*"]）要排除的全局模式（例如，\["/admin/\*\*"]） ```python Extract All URLs theme={null} 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, ...} ``` ```python Filter URLs theme={null} # 仅获取博客 URL result = asyncio.run(extract_urls.ainvoke({ "url": "https://example.com", "include_urls": ["/blog/**"], "exclude_urls": ["/admin/**", "/private/**"], "top_n": 50 })) ``` ### crawl\_website 通过跟随链接自主发现并抓取整个网站。爬取的起始 URL 要爬取的最大页面数要包含的全局模式（例如，\["/\*\*"] 表示全部）要排除的全局模式（例如，\["/admin/\*\*"]）从 start\_url 开始的最大爬取深度包括外部 URL ```python Crawl Website theme={null} 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", ...} ``` ```python With Filters theme={null} # 使用 URL 过滤器进行爬取 result = asyncio.run(crawl_website.ainvoke({ "start_url": "https://example.com", "max_pages": 200, "include_urls": ["/**"], "exclude_urls": ["/admin/**", "/private/**"], "max_depth": 3 })) ``` ## LangChain 代理集成构建可以搜索和抓取网络的智能代理： ```python Basic Agent theme={null} 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) ``` ## LangGraph 集成使用 LangGraph 构建复杂的多步骤工作流： ```python Research Agent theme={null} 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": ""}] } }) ``` ## 高级用例 ### 数据丰富使用网络信息丰富电子表格数据： ```python theme={null} from langchain_olostep import answer_question companies = ["Stripe", "Shopify", "Square"] for company in companies: result = answer_question.invoke({ "task": f"Find information about {company}", "json_schema": { "ceo": "", "headquarters": "", "employee_count": "", "latest_funding": "" } }) print(f"{company}: {result}") ``` ### 电商产品抓取使用专门的解析器抓取产品数据： ```python theme={null} from langchain_olostep import scrape_website # 抓取亚马逊产品 result = scrape_website.invoke({ "url": "https://www.amazon.com/dp/PRODUCT_ID", "parser": "@olostep/amazon-product", "format": "json" }) # 返回结构化的产品数据：价格、标题、评分等 ``` ### SEO 审核分析整个网站的 SEO： ```python theme={null} from langchain_olostep import extract_urls, scrape_batch import json # 1. 发现所有页面 urls_result = extract_urls.invoke({ "url": "https://yoursite.com", "top_n": 1000 }) # 2. 抓取所有页面 urls = json.loads(urls_result)["urls"] batch_result = scrape_batch.invoke({ "urls": urls, "format": "html" }) ``` ### 文档抓取爬取并提取文档： ```python theme={null} from langchain_olostep import crawl_website # 爬取整个文档站点 result = crawl_website.invoke({ "start_url": "https://docs.example.com", "max_pages": 500, "include_urls": ["/docs/**"], "exclude_urls": ["/api/**", "/v1/**"] }) ``` ## 专门解析器 Olostep 提供了针对热门网站的预构建解析器： * `@olostep/google-search` - Google 搜索结果在 `parser` 参数中使用它们： ```python theme={null} scrape_website.invoke({ "url": "https://www.google.com/search?q=alexander+the+great&gl=us&hl=en", "parser": "@olostep/google-search" }) ``` ## 错误处理 ```python theme={null} from langchain_core.exceptions import LangChainException try: result = await scrape_website.ainvoke({ "url": "https://example.com" }) except LangChainException as e: print(f"Scraping failed: {e}") ``` ## 最佳实践当抓取超过 3-5 个 URL 时，使用 `scrape_batch` 而不是多个 `scrape_website` 调用。批处理速度更快且更具成本效益。对于 JavaScript 密集型网站，使用 `wait_before_scraping` 参数（通常为 2000-5000 毫秒）。这确保动态内容完全加载。对于热门网站（如亚马逊、LinkedIn、Google），使用我们的预构建解析器自动获取结构化数据。使用 `extract_urls` 或 `crawl_website` 时，使用全局模式专注于相关页面，避免不必要的处理。对于速率限制错误，实施指数退避。API 会自动处理大多数速率限制。 ## 支持 * **PyPI 包**: [langchain-olostep](https://pypi.org/project/langchain-olostep/) * **文档**: [docs.olostep.com](https://docs.olostep.com) * **问题**: [GitHub Issues](https://github.com/olostep/langchain-olostep/issues) * **电子邮件**: [info@olostep.com](mailto:info@olostep.com) ## 相关资源了解抓取端点了解批处理端点了解答案端点了解地图端点了解爬取端点探索 Python SDK LangChain 平台