|
|
import os |
|
|
import json |
|
|
import requests |
|
|
from typing import Union, List |
|
|
from qwen_agent.tools.base import BaseTool, register_tool |
|
|
from concurrent.futures import ThreadPoolExecutor |
|
|
import http.client |
|
|
|
|
|
|
|
|
SERPER_KEY=os.environ.get('SERPER_KEY_ID') |
|
|
|
|
|
|
|
|
@register_tool("google_scholar", allow_overwrite=True) |
|
|
class Scholar(BaseTool): |
|
|
name = "google_scholar" |
|
|
description = "Leverage Google Scholar to retrieve relevant information from academic publications. Accepts multiple queries." |
|
|
parameters = { |
|
|
"type": "object", |
|
|
"properties": { |
|
|
"query": { |
|
|
"type": "array", |
|
|
"items": {"type": "string", "description": "The search query."}, |
|
|
"minItems": 1, |
|
|
"description": "The list of search queries for Google Scholar." |
|
|
}, |
|
|
}, |
|
|
"required": ["query"], |
|
|
} |
|
|
|
|
|
def google_scholar_with_serp(self, query: str): |
|
|
conn = http.client.HTTPSConnection("google.serper.dev") |
|
|
payload = json.dumps({ |
|
|
"q": query, |
|
|
}) |
|
|
headers = { |
|
|
'X-API-KEY': SERPER_KEY, |
|
|
'Content-Type': 'application/json' |
|
|
} |
|
|
for i in range(5): |
|
|
try: |
|
|
conn.request("POST", "/scholar", payload, headers) |
|
|
res = conn.getresponse() |
|
|
break |
|
|
except Exception as e: |
|
|
print(e) |
|
|
if i == 4: |
|
|
return f"Google Scholar Timeout, return None, Please try again later." |
|
|
continue |
|
|
|
|
|
|
|
|
data = res.read() |
|
|
|
|
|
results = json.loads(data.decode("utf-8")) |
|
|
try: |
|
|
if "organic" not in results: |
|
|
raise Exception(f"No results found for query: '{query}'. Use a less specific query.") |
|
|
|
|
|
web_snippets = list() |
|
|
idx = 0 |
|
|
if "organic" in results: |
|
|
for page in results["organic"]: |
|
|
idx += 1 |
|
|
date_published = "" |
|
|
if "year" in page: |
|
|
date_published = "\nDate published: " + str(page["year"]) |
|
|
|
|
|
publicationInfo = "" |
|
|
if "publicationInfo" in page: |
|
|
publicationInfo = "\npublicationInfo: " + page["publicationInfo"] |
|
|
|
|
|
snippet = "" |
|
|
if "snippet" in page: |
|
|
snippet = "\n" + page["snippet"] |
|
|
|
|
|
link_info = "no available link" |
|
|
if "pdfUrl" in page: |
|
|
link_info = "pdfUrl: " + page["pdfUrl"] |
|
|
|
|
|
citedBy = "" |
|
|
if "citedBy" in page: |
|
|
citedBy = "\ncitedBy: " + str(page["citedBy"]) |
|
|
|
|
|
redacted_version = f"{idx}. [{page['title']}]({link_info}){publicationInfo}{date_published}{citedBy}\n{snippet}" |
|
|
|
|
|
redacted_version = redacted_version.replace("Your browser can't play this video.", "") |
|
|
web_snippets.append(redacted_version) |
|
|
|
|
|
content = f"A Google scholar for '{query}' found {len(web_snippets)} results:\n\n## Scholar Results\n" + "\n\n".join(web_snippets) |
|
|
return content |
|
|
except: |
|
|
return f"No results found for '{query}'. Try with a more general query." |
|
|
|
|
|
|
|
|
def call(self, params: Union[str, dict], **kwargs) -> str: |
|
|
|
|
|
try: |
|
|
params = json.loads(params) |
|
|
params = self._verify_json_format_args(params) |
|
|
query = params["query"] |
|
|
except: |
|
|
return "[google_scholar] Invalid request format: Input must be a JSON object containing 'query' field" |
|
|
|
|
|
if isinstance(query, str): |
|
|
response = self.google_scholar_with_serp(query) |
|
|
else: |
|
|
assert isinstance(query, List) |
|
|
with ThreadPoolExecutor(max_workers=3) as executor: |
|
|
|
|
|
response = list(executor.map(self.google_scholar_with_serp, query)) |
|
|
response = "\n=======\n".join(response) |
|
|
return response |
