""" title: Web Search using LangSearch API author: Rio Zhao Refer: https://langsearch.com/ LangSearch provides web search API for FREE! version: 0.5.0 license: MIT """ import os import requests import json from pydantic import BaseModel, Field import asyncio from typing import Callable, Any from urllib.parse import urlparse class EventEmitter: def __init__(self, event_emitter: Callable[[dict], Any] = None): self.event_emitter = event_emitter async def emit(self, description="Unknown State", status="in_progress", done=False): if self.event_emitter: await self.event_emitter( { "type": "status", "data": { "status": status, "description": description, "done": done, }, } ) class Tools: class Valves(BaseModel): LANGSEARCH_API_ENDPOINT: str = Field( default="https://api.langsearch.com/v1/web-search", description="LangSearch API endpoint", ) LANGSEARCH_API_KEY: str = Field( default="YOUR_API_KEY", description="API key from LangSearch Dashboard" ) SEARCH_COUNT: int = Field( default=5, ge=1, le=10, description="Number of results to return (1-10)" ) FRESHNESS: str = Field( default="noLimit", description="Time filter: oneDay/oneWeek/oneMonth/oneYear/noLimit", ) INCLUDE_SUMMARY: bool = Field( default=True, description="Whether to include full text summaries" ) CITATION_LINKS: bool = Field( default=True, description="Include citations with metadata" ) def __init__(self): self.valves = self.Valves() async def search_web( self, query: str, __event_emitter__: Callable[[dict], Any] = None ) -> str: """ 使用LangSearch Web Search API进行网页搜索 """ emitter = EventEmitter(__event_emitter__) await emitter.emit(f"Sending search request: {query}") headers = { "Authorization": f"Bearer {self.valves.LANGSEARCH_API_KEY}", "Content-Type": "application/json", } payload = { "query": query, "freshness": self.valves.FRESHNESS, "summary": self.valves.INCLUDE_SUMMARY, "count": self.valves.SEARCH_COUNT, } try: # 发送搜索请求 await emitter.emit("正在发送搜索请求...") response = requests.post( self.valves.LANGSEARCH_API_ENDPOINT, headers=headers, json=payload, timeout=30, ) response.raise_for_status() data = response.json() search_results = data.get("data", {}).get("webPages", {}).get("value", []) # 处理搜索结果 processed_results = [] for idx, result in enumerate(search_results, 1): processed = { "title": result.get("name", ""), "url": result.get("url", ""), "snippet": result.get("snippet", ""), "summary": result.get("summary", ""), "display_url": result.get("displayUrl", ""), "domain": self._get_domain(result.get("url", "")), } # 发送引用信息 if self.valves.CITATION_LINKS and __event_emitter__: await __event_emitter__( { "type": "citation", "data": { "document": [ processed["summary"] or processed["snippet"] ], "metadata": [ { "source": processed["url"], "title": processed["title"], "domain": processed["domain"], } ], }, } ) processed_results.append(processed) # 构建最终响应 output = { "total_results": len(processed_results), "results": processed_results, "search_metadata": { "query": query, "freshness": self.valves.FRESHNESS, "summary_enabled": self.valves.INCLUDE_SUMMARY, }, } await emitter.emit( status="complete", description=f"Found {len(processed_results)} relevant results", done=True, ) return json.dumps(output, ensure_ascii=False, indent=2) except requests.exceptions.RequestException as e: error_info = { "error_type": "API Request Error", "message": str(e), "payload": payload, } await emitter.emit( status="error", description=f"搜索请求失败: {str(e)}", done=True ) return json.dumps(error_info, ensure_ascii=False) except json.JSONDecodeError: error_info = { "error_type": "Invalid Response", "message": "Invalid JSON response from API", } await emitter.emit( status="error", description="API返回数据格式错误", done=True ) return json.dumps(error_info, ensure_ascii=False) def _get_domain(self, url: str) -> str: """提取域名""" try: parsed = urlparse(url) return parsed.netloc except: return ""