File size: 5,033 Bytes
73152bb
c02d868
ae12037
a9e9116
667e88a
ae12037
c02d868
ae12037
54b69c4
 
 
49b6791
54b69c4
 
 
 
 
 
 
 
49b6791
54b69c4
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
a9e9116
ae12037
54b69c4
 
 
 
ae12037
 
a9e9116
 
 
 
 
73152bb
a9e9116
54b69c4
 
 
 
 
a9e9116
 
54b69c4
a9e9116
 
 
54b69c4
 
 
a9e9116
 
 
 
 
 
 
 
 
 
 
 
 
 
54b69c4
 
a9e9116
54b69c4
 
427cae4
54b69c4
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
import os
import requests
import urllib.parse
from bs4 import BeautifulSoup

class DuckDuckGoAgent:
    def __init__(self):
        print("DuckDuckGoAgent initialized.")
        self.headers = {
            "User-Agent": "Mozilla/5.0"
        }

        # Support for multiple model backends
        self.supported_models = {
            "huggingface": self.call_huggingface_llm,
            # You can easily extend this dictionary to support:
            # "openai": self.call_openai_model,
            # "lite_llm": self.call_litellm_model,
            # "custom_server": self.call_custom_model,
        }

        self.default_model = "huggingface"
        self.model_config = {
            "huggingface": {
                "api_key": os.getenv("HF_API_TOKEN"),
                "model_name": "mistralai/Mistral-7B-Instruct-v0.1"
            }
        }

    def __call__(self, question: str) -> str:
        """
        Main method to process a question. It first tries DuckDuckGo,
        then scraping, and finally uses a language model if needed.
        """
        print(f"Agent received question: {question[:50]}...")
        answer = self.get_duckduckgo_answer(question)
        print(f"Agent returning answer: {answer}")
        return answer.strip()

    def get_duckduckgo_answer(self, query: str) -> str:
        """
        Attempt to get an answer from the DuckDuckGo API.
        If no abstract text is found, fall back to scraping.
        """
        search_query = urllib.parse.quote(query)
        url = f"https://api.duckduckgo.com/?q={search_query}&format=json&no_html=1&skip_disambig=1"

        try:
            response = requests.get(url, timeout=10)
            if response.status_code == 200:
                data = response.json()
                if 'AbstractText' in data and data['AbstractText']:
                    return data['AbstractText'][:200]
                else:
                    print("No abstract found, falling back to scraping.")
                    return self.scrape_duckduckgo(query)
            else:
                print(f"DuckDuckGo API failed with status: {response.status_code}")
                return self.scrape_duckduckgo(query)
        except Exception as e:
            print(f"Error contacting DuckDuckGo API: {e}")
            return self.scrape_duckduckgo(query)

    def scrape_duckduckgo(self, query: str) -> str:
        """
        Fallback to scraping DuckDuckGo search results if API fails or no abstract found.
        """
        print("Using fallback: scraping HTML results.")
        try:
            response = requests.post(
                "https://html.duckduckgo.com/html/",
                data={"q": query},
                headers=self.headers,
                timeout=10
            )
            soup = BeautifulSoup(response.text, "html.parser")
            snippets = soup.select(".result__snippet")
            for s in snippets:
                text = s.get_text().strip()
                if text:
                    return text[:200]
            print("No useful snippets found, falling back to language model.")
            return self.call_model_backend(query)
        except Exception as e:
            print(f"Error scraping DuckDuckGo: {e}")
            return self.call_model_backend(query)

    def call_model_backend(self, prompt: str) -> str:
        """
        Dispatch to the selected LLM backend.
        """
        if self.default_model in self.supported_models:
            return self.supported_models[self.default_model](prompt)
        return "No valid model backend configured."

    def call_huggingface_llm(self, prompt: str) -> str:
        """
        Call Hugging Face Inference API as fallback LLM.
        """
        config = self.model_config.get("huggingface", {})
        api_key = config.get("api_key")
        model = config.get("model_name")

        if not api_key or not model:
            return "Error: Hugging Face API Token or model not configured."

        url = f"https://api-inference.huggingface.co/models/{model}"
        headers = {
            "Authorization": f"Bearer {api_key}",
            "Content-Type": "application/json"
        }
        payload = {
            "inputs": prompt,
            "parameters": {
                "max_new_tokens": 200,
                "temperature": 0.7
            }
        }

        try:
            response = requests.post(url, headers=headers, json=payload, timeout=30)
            response.raise_for_status()
            output = response.json()
            if isinstance(output, list) and "generated_text" in output[0]:
                return output[0]["generated_text"].strip()[:200]
            elif isinstance(output, dict) and "error" in output:
                return f"HF LLM error: {output['error']}"
            else:
                return "No response generated from Hugging Face LLM."
        except Exception as e:
            print(f"Error contacting Hugging Face LLM: {e}")
            return "Error contacting Hugging Face model."