Update tools.py

tools.py

@@ -1,257 +1,103 @@
-from smolagents import DuckDuckGoSearchTool
-from smolagents import Tool
-import random
-from huggingface_hub import list_models
-
-
-# Initialize the DuckDuckGo search tool
-#search_tool = DuckDuckGoSearchTool()
-
-
-class WeatherInfoTool(Tool):
-    name = "weather_info"
-    description = "Fetches dummy weather information for a given location."
-    inputs = {
-        "location": {
-            "type": "string",
-            "description": "The location to get weather information for."
-        }
-    }
-    output_type = "string"
-
-    def forward(self, location: str):
-        # Dummy weather data
-        weather_conditions = [
-            {"condition": "Rainy", "temp_c": 15},
-            {"condition": "Clear", "temp_c": 25},
-            {"condition": "Windy", "temp_c": 20}
-        ]
-        # Randomly select a weather condition
-        data = random.choice(weather_conditions)
-        return f"Weather in {location}: {data['condition']}, {data['temp_c']}°C"
-
-class HubStatsTool(Tool):
-    name = "hub_stats"
-    description = "Fetches the most downloaded model from a specific author on the Hugging Face Hub."
-    inputs = {
-        "author": {
-            "type": "string",
-            "description": "The username of the model author/organization to find models from."
-        }
-    }
-    output_type = "string"
-
-    def forward(self, author: str):
-        try:
-            # List models from the specified author, sorted by downloads
-            models = list(list_models(author=author, sort="downloads", direction=-1, limit=1))
-            
-            if models:
-                model = models[0]
-                return f"The most downloaded model by {author} is {model.id} with {model.downloads:,} downloads."
-            else:
-                return f"No models found for author {author}."
-        except Exception as e:
-            return f"Error fetching models for {author}: {str(e)}"
-
-
-# from typing import TypedDict, List, Optional, Annotated
-# from langchain_core.messages import BaseMessage
-# from langgraph.graph import StateGraph, END
-# from langchain_core.prompts import ChatPromptTemplate
-# from langchain_openai import ChatOpenAI
-# from retriever import get_retriever
-# import json
+# from smolagents import DuckDuckGoSearchTool
+# from smolagents import Tool
+# import random
+# from huggingface_hub import list_models
 
-# # 定义状态对象
-# class GraphState(TypedDict):
-#     question: str
-#     documents: List[str]
-#     answer: str
-#     verification: Annotated[Optional[dict], "验证结果"]
-#     retries: Annotated[int, "剩余重试次数"]
-#     feedback: Annotated[Optional[str], "前次验证的反馈"]
-#     history: Annotated[List[dict], "执行历史记录"]
 
-# # 初始化检索器和模型
-# retriever = get_retriever()
-# llm = ChatOpenAI(model="gpt-3.5-turbo", temperature=0.7)
+# # Initialize the DuckDuckGo search tool
+# #search_tool = DuckDuckGoSearchTool()
 
-# def retrieve(state: GraphState):
-#     """检索文档节点"""
-#     history = state["history"]
-#     history.append({"step": "检索", "status": "开始"})
-    
-#     question = state["question"]
-#     documents = retriever.get_relevant_documents(question)
-#     doc_contents = [doc.page_content for doc in documents]
-    
-#     history.append({
-#         "step": "检索", 
-#         "status": "完成",
-#         "documents": doc_contents
-#     })
-    
-#     return {"documents": doc_contents, "history": history}
 
-# def generate(state: GraphState):
-#     """生成答案节点"""
-#     history = state["history"]
-#     history.append({"step": "生成", "status": "开始"})
-    
-#     question = state["question"]
-#     documents = state["documents"]
-#     feedback = state.get("feedback", "")
-    
-#     # 构建提示词
-#     prompt = ChatPromptTemplate.from_messages([
-#         ("system", "你是一个专业助手，基于以下上下文回答问题。如果上下文不足，请说明。{feedback}"),
-#         ("human", "问题：{question}\n上下文：{context}")
-#     ])
-    
-#     chain = prompt | llm
-#     context = "\n\n".join(documents)
-#     response = chain.invoke({
-#         "question": question,
-#         "context": context,
-#         "feedback": feedback
-#     })
-    
-#     history.append({
-#         "step": "生成", 
-#         "status": "完成",
-#         "answer": response.content
-#     })
-    
-#     return {"answer": response.content, "history": history}
-
-# def verify(state: GraphState):
-#     """验证答案节点"""
-#     history = state["history"]
-#     history.append({"step": "验证", "status": "开始"})
-    
-#     question = state["question"]
-#     answer = state["answer"]
-#     documents = state["documents"]
-    
-#     # 验证提示词
-#     prompt = ChatPromptTemplate.from_messages([
-#         ("system", "评估答案是否符合以下标准：\n"
-#          "1. 是否基于提供的上下文\n"
-#          "2. 是否完整回答问题\n"
-#          "3. 是否包含幻觉信息\n\n"
-#          "返回JSON格式：{\"valid\": boolean, \"feedback\": string}"),
-#         ("human", "问题：{question}\n答案：{answer}\n上下文：{context}")
-#     ])
-    
-#     chain = prompt | llm
-#     context = "\n\n".join(documents)
-#     result = chain.invoke({
-#         "question": question,
-#         "answer": answer,
-#         "context": context
-#     })
-    
-#     try:
-#         # 尝试解析JSON输出
-#         verification = json.loads(result.content)
-#     except:
-#         # 如果解析失败，使用默认值
-#         verification = {"valid": False, "feedback": "验证失败: 无法解析验证结果"}
-    
-#     history.append({
-#         "step": "验证", 
-#         "status": "完成",
-#         "verification": verification
-#     })
-    
-#     return {"verification": verification, "history": history}
-
-# def should_retry(state: GraphState):
-#     """决定是否重试的条件函数"""
-#     history = state["history"]
-    
-#     if state["verification"].get("valid", False):
-#         history.append({"step": "决策", "action": "验证通过，结束流程"})
-#         return "end"
-#     elif state["retries"] > 0:
-#         history.append({
-#             "step": "决策", 
-#             "action": f"验证失败，剩余重试次数：{state['retries']}，将重试"
-#         })
-#         return "retry"
-#     else:
-#         history.append({"step": "决策", "action": "重试次数用尽，结束流程"})
-#         return "end"
-
-# def prepare_retry(state: GraphState):
-#     """准备重试节点"""
-#     history = state["history"]
-#     history.append({"step": "准备重试", "status": "开始"})
-    
-#     feedback = state["verification"].get("feedback", "需要改进答案")
-    
-#     history.append({
-#         "step": "准备重试", 
-#         "status": "完成",
-#         "feedback": feedback
-#     })
-    
-#     return {
-#         "feedback": feedback,
-#         "retries": state["retries"] - 1,
-#         "history": history
+# class WeatherInfoTool(Tool):
+#     name = "weather_info"
+#     description = "Fetches dummy weather information for a given location."
+#     inputs = {
+#         "location": {
+#             "type": "string",
+#             "description": "The location to get weather information for."
+#         }
 #     }
+#     output_type = "string"
+
+#     def forward(self, location: str):
+#         # Dummy weather data
+#         weather_conditions = [
+#             {"condition": "Rainy", "temp_c": 15},
+#             {"condition": "Clear", "temp_c": 25},
+#             {"condition": "Windy", "temp_c": 20}
+#         ]
+#         # Randomly select a weather condition
+#         data = random.choice(weather_conditions)
+#         return f"Weather in {location}: {data['condition']}, {data['temp_c']}°C"
+
+# class HubStatsTool(Tool):
+#     name = "hub_stats"
+#     description = "Fetches the most downloaded model from a specific author on the Hugging Face Hub."
+#     inputs = {
+#         "author": {
+#             "type": "string",
+#             "description": "The username of the model author/organization to find models from."
+#         }
+#     }
+#     output_type = "string"
 
-# # 构建工作流
-# workflow = StateGraph(GraphState)
-
-# # 添加节点
-# workflow.add_node("retrieve", retrieve)
-# workflow.add_node("generate", generate)
-# workflow.add_node("verify", verify)
-# workflow.add_node("prepare_retry", prepare_retry)
-
-# # 设置入口点
-# workflow.set_entry_point("retrieve")
+#     def forward(self, author: str):
+#         try:
+#             # List models from the specified author, sorted by downloads
+#             models = list(list_models(author=author, sort="downloads", direction=-1, limit=1))
+            
+#             if models:
+#                 model = models[0]
+#                 return f"The most downloaded model by {author} is {model.id} with {model.downloads:,} downloads."
+#             else:
+#                 return f"No models found for author {author}."
+#         except Exception as e:
+#             return f"Error fetching models for {author}: {str(e)}"
 
-# # 添加边
-# workflow.add_edge("retrieve", "generate")
-# workflow.add_edge("generate", "verify")
-# workflow.add_conditional_edges(
-#     "verify",
-#     should_retry,
-#     {
-#         "end": END,
-#         "retry": "prepare_retry"
-#     }
-# )
-# workflow.add_edge("prepare_retry", "retrieve")
 
-# # 编译工作流
-# app = workflow.compile()
+from langchain.tools import Tool
+from huggingface_hub import list_models
+import random
 
-# def run_agentic_rag(question: str, max_retries: int = 3):
-#     """运行Agentic RAG工作流"""
-#     initial_state = {
-#         "question": question,
-#         "documents": [],
-#         "answer": "",
-#         "verification": None,
-#         "retries": max_retries,
-#         "feedback": "",
-#         "history": [{"step": "初始化", "status": f"开始处理问题: {question}"}]
-#     }
-    
-#     # 执行工作流
-#     final_state = None
-#     for step in app.stream(initial_state):
-#         node, state = next(iter(step.items()))
-#         final_state = state
-    
-#     return {
-#         "answer": final_state["answer"],
-#         "documents": final_state["documents"],
-#         "history": final_state["history"],
-#         "retries_used": max_retries - final_state["retries"]
-#     }
+def get_weather_info(location: str) -> str:
+    """Fetches dummy weather information for a given location."""
+    # Dummy weather data
+    weather_conditions = [
+        {"condition": "Rainy", "temp_c": 15},
+        {"condition": "Clear", "temp_c": 25},
+        {"condition": "Windy", "temp_c": 20}
+    ]
+    # Randomly select a weather condition
+    data = random.choice(weather_conditions)
+    return f"Weather in {location}: {data['condition']}, {data['temp_c']}°C"
+
+# Initialize the tool
+weather_info_tool = Tool(
+    name="get_weather_info",
+    func=get_weather_info,
+    description="Fetches dummy weather information for a given location."
+)
+
+def get_hub_stats(author: str) -> str:
+    """Fetches the most downloaded model from a specific author on the Hugging Face Hub."""
+    try:
+        # List models from the specified author, sorted by downloads
+        models = list(list_models(author=author, sort="downloads", direction=-1, limit=1))
+
+        if models:
+            model = models[0]
+            return f"The most downloaded model by {author} is {model.id} with {model.downloads:,} downloads."
+        else:
+            return f"No models found for author {author}."
+    except Exception as e:
+        return f"Error fetching models for {author}: {str(e)}"
+
+# Initialize the tool
+hub_stats_tool = Tool(
+    name="get_hub_stats",
+    func=get_hub_stats,
+    description="Fetches the most downloaded model from a specific author on the Hugging Face Hub."
+)
+
+# Example usage
+print(hub_stats_tool("facebook")) # Example: Get the most downloaded model by Facebook