sk75's picture
Upload 8 files
1b38924 verified
Raw
History Blame Contribute Delete
15.4 kB
import requests
from typing import Dict, Any
import xml.etree.ElementTree as ET
import database
def search_arxiv_papers(query:str, max_results:int=10) -> Dict[str, Any]:
"""
Arxiv üzerinden kullanıcı sorgusunu arar ve başlık, yazar, özet ve PDF linklerini döner.
Inputs:
query: Kullanıcı sorgusu (Örneğin kullanıcı son 2 yılda çıkan Machine Learning makalelerini ara diyebilir.)
max_results: Dönen sonuçlar içerisinden seçilecek maksimum makale miktarı
Outputs:
Dict : {status: "success" or "error", count: int, papers: List[Dict] | None, message: str | None}
"""
try:
url = f"http://export.arxiv.org/api/query?search_query=all:{query}&start=0&max_results={max_results}"
response = requests.get(url, timeout=10)
root = ET.fromstring(response.text)
ns = {
"atom": "http://www.w3.org/2005/Atom"
}
papers = []
for entry in root.findall("atom:entry", ns):
arxiv_id_elem = entry.find('atom:id', ns)
raw_id = arxiv_id_elem.text if arxiv_id_elem is not None else ""
paper_id = raw_id.split('/abs/')[-1] if '/abs/' in raw_id else raw_id
title = entry.find("atom:title", ns).text.strip()
summary = entry.find("atom:summary", ns).text.strip()
authors = [author.find("atom:name", ns).text for author in entry.findall("atom:author", ns)]
pdf = None
for link in entry.findall("atom:link", ns):
if link.attrib.get("title") == "pdf":
pdf = link.attrib["href"]
published_elem = entry.find('atom:published', ns)
published = published_elem.text[:10] if published_elem is not None else ""
year = published[:4] if published else ""
authors = []
for author in entry.findall('atom:author', ns):
name_elem = author.find('atom:name', ns)
if name_elem is not None and name_elem.text:
authors.append(name_elem.text)
authors_str = ", ".join(authors)
paper_url = f"https://arxiv.org/abs/{paper_id}"
papers.append({
"paper_id": f"arxiv_{paper_id}",
"title": title,
"authors": authors_str,
"summary": summary[:350] + "..." if len(summary) > 350 else summary,
"url": paper_url,
"published_year": year,
"source": "ArXiv"
})
return {"status": "success", "count": len(papers), "papers": papers}
except Exception as e:
return {"status": "error", "message": str(e)}
def search_openalex(query: str, limit: int = 5) -> Dict[str, Any]:
"""OpenAlex üzerinden atıf sayıları (citations) ile zenginleştirilmiş akademik makaleleri getirir.
Inputs:
query: Kullanıcı sorgusu (Örneğin kullanıcı son 2 yılda çıkan Machine Learning makalelerini ara diyebilir.)
limit: Dönen sonuçlar içerisinden seçilecek maksimum makale miktarı
Outputs:
Dict : {status: "success" or "error", count: int, papers: List[Dict] | None, message: str | None}
"""
url = "https://api.openalex.org/works"
params = {
"search": query,
"per_page": limit,
"select": "id,title,publication_year,cited_by_count,authorships,doi,primary_location"
}
response = requests.get(url, params=params, timeout=10)
if response.status_code != 200:
print("OpenAlex API rate limit hatası oluştu, Arxiv API kullanılıyor...")
return search_arxiv_papers(query=query, max_results=limit)
data = response.json()
papers = []
for item in data.get("results", []):
paper_id = item.get("id", "").split("/")[-1]
title = item.get("title", "")
authors = [
author["author"]["display_name"]
for author in item.get("authorships", [])
]
authors_str = ", ".join(authors)
summary = ""
if "abstract" in item:
summary = item["abstract"]
year = item.get("publication_year", "")
citation_count = item.get("cited_by_count", 0)
location = item.get("primary_location", {})
paper_url = (
location.get("landing_page_url")
or item.get("doi")
or item.get("id")
)
papers.append({
"paper_id": f"openalex_{paper_id}",
"title": title,
"authors": authors_str,
"summary": summary[:350] + "..." if len(summary) > 350 else summary,
"citation_count": citation_count,
"published_year": year,
"url": paper_url,
"source": "OpenAlex"
})
return {
"status": "success",
"count": len(papers),
"papers": papers
}
def save_paper_to_library(paper_id: str, title: str, authors: str = "",
summary: str = "", url: str = "", source: str = "ArXiv", citation_count: int = 0,
published_year: str = "", tags: str = "") -> Dict[str, Any]:
"""Seçilen makaleyi SQLite kişisel veritabanına kaydeder.
Inputs:
paper_id: Makalenin benzersiz ID'si
title: Makalenin başlığı
authors: Makalenin yazarları
summary: Makalenin özeti
url: Makalenin URL'si
source: Makalenin kaynağı
citation_count: Makalenin atıf sayısı
published_year: Makalenin yayınlanma yılı
tags: Makalenin etiketleri
Outputs:
Dict : {status: "success" or "error", message: str}
"""
return database.save_paper(
paper_id=paper_id,
title=title,
authors=authors,
summary=summary,
url=url,
source=source,
citation_count=citation_count,
published_year=published_year,
tags=tags
)
def get_saved_library(status_filter: str = "hepsi", tag_filter: str = "", query: str = "") -> Dict[str, Any]:
"""Kayıtlı makale kütüphanesini okur ve listeler.
Inputs:
status_filter: Durum
tag_filter: Etiket
query: Arama sorgusu
Outputs:
Dict : {status: "success" or "error", count: int, papers: List[Dict] | None, message: str | None}
"""
papers = database.get_saved_papers(status_filter=status_filter, tag_filter=tag_filter, query=query)
return {
"status": "success",
"saved_count": len(papers),
"library": papers
}
def update_paper_status_or_note(paper_id: str, status: str = "", notes: str = "", tags: str = "") -> Dict[str, Any]:
"""Kayıtlı bir makalenin okuma durumunu ('unread', 'reading', 'completed') veya kişisel notlarını günceller.
Inputs:
paper_id: Makalenin benzersiz ID'si
status: Makalenin okuma durumu
notes: Makalenin kişisel notları
tags: Makalenin etiketleri
Outputs:
Dict : {status: "success" or "error", message: str}
"""
return database.update_paper_status_or_note(paper_id=paper_id, status=status if status else None, notes=notes if notes else None, tags=tags if tags else None)
def execute_tool(tool_name: str, arguments: Dict[str, Any]) -> Dict[str, Any]:
"""LLM tarafından çağrılan fonksiyonu eşleştirir ve çalıştırır.
Inputs:
tool_name: Fonksiyonun adı
arguments: Fonksiyonun argümanları
Outputs:
Dict : {status: "success" or "error", message: str}
"""
if not isinstance(arguments, dict):
arguments = {}
cleaned_args = {k: v for k, v in arguments.items() if v is not None}
if tool_name == "search_arxiv_papers":
return search_arxiv_papers(**cleaned_args)
elif tool_name == "search_openalex":
return search_openalex(**cleaned_args)
elif tool_name == "save_paper_to_library":
return save_paper_to_library(**cleaned_args)
elif tool_name == "get_saved_library":
return get_saved_library(**cleaned_args)
elif tool_name == "update_paper_status_or_note":
return update_paper_status_or_note(**cleaned_args)
elif tool_name in ["delete_paper", "delete_paper_from_library"]:
return delete_paper(**cleaned_args)
elif tool_name in ["clear_library", "clear_entire_library"]:
return clear_library()
else:
return {"status": "error", "message": f"Bilinmeyen fonksiyon: {tool_name}"}
def delete_paper(paper_id:str) -> Dict[str, Any]:
"""Kayıtlı bir makaleyi veritabanından siler.
Inputs:
paper_id: Makalenin benzersiz ID'si
Outputs:
Dict : {status: "success" or "error", message: str}
"""
return database.delete_paper(paper_id=paper_id)
def clear_library() -> Dict[str, Any]:
"""Kütüphanedeki TÜM makaleleri veritabanından tamamen siler ve temizler.
Inputs:
None
Outputs:
Dict : {status: "success" or "error", message: str}
"""
return database.clear_library()
TOOLS_SCHEMA = [
{
"type": "function",
"function": {
"name": "search_arxiv_papers",
"description": "ArXiv akademik platformundan verilen konu veya kelimelere göre en güncel makaleleri arar ve getirir.",
"parameters": {
"type": "object",
"properties": {
"query": {
"type": "string",
"description": "Aranacak akademik konu veya anahtar kelimeler (örn: 'quantum machine learning', 'LLM tool calling')"
},
"max_results": {
"type": "integer",
"description": "Getirilecek maksimum makale sayısı (Varsayılan 10)",
"default": 10
}
},
"required": ["query"]
}
}
},
{
"type": "function",
"function": {
"name": "search_openalex",
"description": "OpenAlex platformundan makaleleri atıf sayıları (citations) ve yayın yılı bilgisi ile getirir.",
"parameters": {
"type": "object",
"properties": {
"query": {
"type": "string",
"description": "Aranacak makale konusu"
},
"limit": {
"type": "integer",
"description": "Getirilecek makale sayısı",
"default": 5
}
},
"required": ["query"]
}
}
},
{
"type": "function",
"function": {
"name": "save_paper_to_library",
"description": "Arama sonuçlarında bulunan bir makaleyi veritabanındaki kişisel kütüphaneye kaydeder.",
"parameters": {
"type": "object",
"properties": {
"paper_id": {"type": "string", "description": "Makalenin benzersiz kimliği (örn: arxiv_2303.08774)"},
"title": {"type": "string", "description": "Makale başlığı"},
"authors": {"type": "string", "description": "Yazarlar", "default": ""},
"summary": {"type": "string", "description": "Makale özeti", "default": ""},
"url": {"type": "string", "description": "Makale web/PDF bağlantısı", "default": ""},
"source": {"type": "string", "description": "Kaynak platform (ArXiv, OpenAlex)", "default": "ArXiv"},
"citation_count": {"type": "integer", "description": "Atıf sayısı", "default": 0},
"published_year": {"type": "string", "description": "Yayın yılı", "default": ""},
"tags": {"type": "string", "description": "Kullanıcı etiketleri", "default": ""}
},
"required": ["paper_id", "title"]
}
}
},
{
"type": "function",
"function": {
"name": "get_saved_library",
"description": "Veritabanına önceden kaydedilmiş makaleleri ve okuma listesini getirir. Filtre kullanılmayacaksa query ve tag_filter parametrelerine boş metin '' verilmeli veya dahil edilmemelidir.",
"parameters": {
"type": "object",
"properties": {
"status_filter": {
"type": "string",
"description": "Filtreleme durumu: 'hepsi', 'unread', 'reading', 'completed'",
"default": "hepsi"
},
"tag_filter": {
"type": "string",
"description": "Etikete göre filtreleme. Özel filtre yoksa '' (boş metin) kullanın.",
"default": ""
},
"query": {
"type": "string",
"description": "Başlık veya özet içinde kelime araması. Özel filtre yoksa '' (boş metin) kullanın.",
"default": ""
}
}
}
}
},
{
"type": "function",
"function": {
"name": "update_paper_status_or_note",
"description": "Kütüphanede kayıtlı bir makalenin okuma durumunu ('unread', 'reading', 'completed') günceller veya kişisel not/etiket ekler.",
"parameters": {
"type": "object",
"properties": {
"paper_id": {"type": "string", "description": "Güncellenecek makalenin kimliği veya ID'si"},
"status": {"type": "string", "description": "Yeni okuma durumu: 'unread', 'reading', 'completed'", "default": ""},
"notes": {"type": "string", "description": "Kullanıcının makaleye eklemek istediği özel not", "default": ""},
"tags": {"type": "string", "description": "Güncellenecek etiketler", "default": ""}
},
"required": ["paper_id"]
}
}
},
{
"type": "function",
"function": {
"name": "delete_paper",
"description": "Kütüphanede kayıtlı bir makaleyi veritabanından siler.",
"parameters": {
"type": "object",
"properties": {
"paper_id": {"type": "string", "description": "Silinecek makalenin kimliği veya ID'si"}
},
"required": ["paper_id"]
}
}
},
{
"type": "function",
"function": {
"name": "clear_library",
"description": "Kütüphanedeki TÜM makaleleri veritabanından tamamen siler ve temizler.",
"parameters": {
"type": "object",
"properties": {},
"required": []
}
}
}
]