| 1 | import asyncio |
| 2 | from helpers import dotenv, perplexity_search, duckduckgo_search |
| 3 | from plugins._memory.helpers.memory import Memory |
| 4 | from plugins._memory.tools.memory_load import DEFAULT_THRESHOLD as DEFAULT_MEMORY_THRESHOLD |
| 5 | |
| 6 | from helpers.tool import Tool, Response |
| 7 | from plugins._document_query.helpers.document_query import DocumentQueryHelper |
| 8 | |
| 9 | SEARCH_ENGINE_RESULTS = 10 |
| 10 | |
| 11 | |
| 12 | class Knowledge(Tool): |
| 13 | async def execute(self, question="", **kwargs): |
| 14 | if not question: |
| 15 | question = kwargs.get("query", "") |
| 16 | if not question: |
| 17 | return Response(message="No question provided", break_loop=False) |
| 18 | |
| 19 | # Create tasks for all search methods |
| 20 | tasks = [ |
| 21 | self.searxng_search(question), |
| 22 | # self.perplexity_search(question), |
| 23 | # self.duckduckgo_search(question), |
| 24 | self.mem_search_enhanced(question), |
| 25 | ] |
| 26 | |
| 27 | # Run all tasks concurrently |
| 28 | results = await asyncio.gather(*tasks, return_exceptions=True) |
| 29 | |
| 30 | # perplexity_result, duckduckgo_result, memory_result = results |
| 31 | searxng_result, memory_result = results |
| 32 | |
| 33 | # enrich results with qa |
| 34 | searxng_result = await self.searxng_document_qa(searxng_result, question) |
| 35 | |
| 36 | # Handle exceptions and format results |
| 37 | searxng_result = self.format_result_searxng(searxng_result, "Search Engine") |
| 38 | memory_result = self.format_result(memory_result, "Memory") |
| 39 | |
| 40 | msg = self.agent.read_prompt( |
| 41 | "fw.knowledge_tool.response.md", |
| 42 | # online_sources = ((perplexity_result + "\n\n") if perplexity_result else "") + str(duckduckgo_result), |
| 43 | online_sources=((searxng_result + "\n\n") if searxng_result else ""), |
| 44 | memory=memory_result, |
| 45 | ) |
| 46 | |
| 47 | await self.agent.handle_intervention( |
| 48 | msg |
| 49 | ) # wait for intervention and handle it, if paused |
| 50 | |
| 51 | return Response(message=msg, break_loop=False) |
| 52 | |
| 53 | async def perplexity_search(self, question): |
| 54 | if dotenv.get_dotenv_value("API_KEY_PERPLEXITY"): |
| 55 | return await asyncio.to_thread( |
| 56 | perplexity_search.perplexity_search, question |
| 57 | ) |
| 58 | else: |
| 59 | PrintStyle.hint( |
| 60 | "No API key provided for Perplexity. Skipping Perplexity search." |
| 61 | ) |
| 62 | self.agent.context.log.log( |
| 63 | type="hint", |
| 64 | content="No API key provided for Perplexity. Skipping Perplexity search.", |
| 65 | ) |
| 66 | return None |
| 67 | |
| 68 | async def duckduckgo_search(self, question): |
| 69 | return await asyncio.to_thread(duckduckgo_search.search, question) |
| 70 | |
| 71 | async def searxng_search(self, question): |
| 72 | return await searxng(question) |
| 73 | |
| 74 | async def searxng_document_qa(self, result, query): |
| 75 | if isinstance(result, Exception) or not query or not result or not result["results"]: |
| 76 | return result |
| 77 | |
| 78 | result["results"] = result["results"][:SEARCH_ENGINE_RESULTS] |
| 79 | |
| 80 | tasks = [] |
| 81 | helper = DocumentQueryHelper(self.agent) |
| 82 | |
| 83 | for index, item in enumerate(result["results"]): |
| 84 | tasks.append(helper.document_qa(item["url"], [query])) |
| 85 | |
| 86 | task_results = list(await asyncio.gather(*tasks, return_exceptions=True)) |
| 87 | |
| 88 | for index, item in enumerate(result["results"]): |
| 89 | if isinstance(task_results[index], BaseException): |
| 90 | continue |
| 91 | found, qa = task_results[index] # type: ignore |
| 92 | if not found: |
| 93 | continue |
| 94 | result["results"][index]["qa"] = qa |
| 95 | |
| 96 | return result |
| 97 | |
| 98 | async def mem_search(self, question: str): |
| 99 | db = await Memory.get(self.agent) |
| 100 | docs = await db.search_similarity_threshold( |
| 101 | query=question, limit=5, threshold=DEFAULT_MEMORY_THRESHOLD |
| 102 | ) |
| 103 | text = Memory.format_docs_plain(docs) |
| 104 | return "\n\n".join(text) |
| 105 | |
| 106 | async def mem_search_enhanced(self, question: str): |
| 107 | """ |
| 108 | Enhanced memory search with knowledge source awareness. |
| 109 | Separates and prioritizes knowledge sources vs conversation memories. |
| 110 | """ |
| 111 | try: |
| 112 | db = await Memory.get(self.agent) |
| 113 | |
| 114 | # Search for knowledge sources (knowledge_source=True) |
| 115 | knowledge_docs = await db.search_similarity_threshold( |
| 116 | query=question, limit=5, threshold=DEFAULT_MEMORY_THRESHOLD, |
| 117 | filter="knowledge_source == True" |
| 118 | ) |
| 119 | |
| 120 | # Search for conversation memories (field doesn't exist or is not True) |
| 121 | conversation_docs = await db.search_similarity_threshold( |
| 122 | query=question, limit=5, threshold=DEFAULT_MEMORY_THRESHOLD, |
| 123 | filter="not knowledge_source if 'knowledge_source' in locals() else True" |
| 124 | ) |
| 125 | |
| 126 | # Combine and fallback to lower threshold if needed |
| 127 | all_docs = knowledge_docs + conversation_docs |
| 128 | threshold_note = "" |
| 129 | |
| 130 | # If no results with default threshold, try with lower threshold |
| 131 | if not all_docs: |
| 132 | lower_threshold = DEFAULT_MEMORY_THRESHOLD * 0.8 |
| 133 | knowledge_docs = await db.search_similarity_threshold( |
| 134 | query=question, limit=5, threshold=lower_threshold, |
| 135 | filter="knowledge_source == True" |
| 136 | ) |
| 137 | conversation_docs = await db.search_similarity_threshold( |
| 138 | query=question, limit=5, threshold=lower_threshold, |
| 139 | filter="not knowledge_source if 'knowledge_source' in locals() else True" |
| 140 | ) |
| 141 | all_docs = knowledge_docs + conversation_docs |
| 142 | if all_docs: |
| 143 | threshold_note = f" (threshold: {lower_threshold})" |
| 144 | |
| 145 | if not all_docs: |
| 146 | return await self._get_memory_diagnostics(db, question) |
| 147 | |
| 148 | # Separate knowledge sources from conversation memories |
| 149 | knowledge_sources = knowledge_docs |
| 150 | conversation_memories = conversation_docs |
| 151 | result_parts = [] |
| 152 | |
| 153 | # Add search summary |
| 154 | result_parts.append(f"## 🔍 Search Results for: '{question}'") |
| 155 | result_parts.append(f"**Found:** {len(knowledge_sources)} knowledge sources, {len(conversation_memories)} conversation memories{threshold_note}") |
| 156 | |
| 157 | # Show knowledge sources |
| 158 | if knowledge_sources: |
| 159 | result_parts.append("") |
| 160 | result_parts.append("## 📚 Knowledge Sources:") |
| 161 | for index, doc in enumerate(knowledge_sources): |
| 162 | source_file = doc.metadata.get('source_file', 'Unknown source') |
| 163 | file_type = doc.metadata.get('file_type', '').upper() |
| 164 | area = doc.metadata.get('area', 'main').upper() |
| 165 | |
| 166 | result_parts.append(f"**Source:** {source_file} ({file_type}) [{area}]") |
| 167 | result_parts.append(f"**Content:** {doc.page_content}") |
| 168 | if index < len(knowledge_sources) - 1: |
| 169 | result_parts.append("-" * 80) |
| 170 | |
| 171 | # Show conversation memories |
| 172 | if conversation_memories: |
| 173 | if knowledge_sources: |
| 174 | result_parts.append("") |
| 175 | result_parts.append("## 💭 Related Experience:") |
| 176 | for index, doc in enumerate(conversation_memories): |
| 177 | timestamp = doc.metadata.get('timestamp', 'Unknown time') |
| 178 | area = doc.metadata.get('area', 'main').upper() |
| 179 | consolidation_action = doc.metadata.get('consolidation_action', '') |
| 180 | |
| 181 | metadata_info = f"{timestamp} [{area}]" |
| 182 | if consolidation_action: |
| 183 | metadata_info += f" (consolidated: {consolidation_action})" |
| 184 | |
| 185 | result_parts.append(f"**Experience:** {metadata_info}") |
| 186 | result_parts.append(f"**Content:** {doc.page_content}") |
| 187 | if index < len(conversation_memories) - 1: |
| 188 | result_parts.append("-" * 80) |
| 189 | |
| 190 | return "\n".join(result_parts) |
| 191 | |
| 192 | except Exception as e: |
| 193 | handle_error(e) |
| 194 | return f"Memory search failed: {str(e)}" |
| 195 | |
| 196 | async def _get_memory_diagnostics(self, db, query: str): |
| 197 | """Provide memory diagnostics when no search results are found.""" |
| 198 | try: |
| 199 | # Get sample of all documents to see what's in memory |
| 200 | sample_docs = await db.search_similarity_threshold( |
| 201 | query="test", limit=20, threshold=0.0 |
| 202 | ) |
| 203 | |
| 204 | if not sample_docs: |
| 205 | return f"## 🔍 No Results for: '{query}'\n**Memory database appears to be empty.**" |
| 206 | |
| 207 | # Analyze what's in memory |
| 208 | area_counts: dict[str, int] = {} |
| 209 | knowledge_count = 0 |
| 210 | |
| 211 | for doc in sample_docs: |
| 212 | area = doc.metadata.get('area', 'unknown') |
| 213 | area_counts[area] = area_counts.get(area, 0) + 1 |
| 214 | if doc.metadata.get('knowledge_source', False): |
| 215 | knowledge_count += 1 |
| 216 | |
| 217 | result_parts = [ |
| 218 | f"## 🔍 No Results for: '{query}'", |
| 219 | f"**Database contains:** {len(sample_docs)} total documents", |
| 220 | f"**Areas:** {', '.join([f'{area.upper()}: {count}' for area, count in area_counts.items()])}", |
| 221 | f"**Knowledge sources:** {knowledge_count} documents", |
| 222 | "", |
| 223 | "**Suggestions:**", |
| 224 | "- Try different or more general search terms", |
| 225 | "- Check if the information was recently memorized", |
| 226 | f"- Current search threshold: {DEFAULT_MEMORY_THRESHOLD}" |
| 227 | ] |
| 228 | |
| 229 | return "\n".join(result_parts) |
| 230 | |
| 231 | except Exception as e: |
| 232 | return f"Memory diagnostics failed: {str(e)}" |
| 233 | |
| 234 | def format_result(self, result, source): |
| 235 | if isinstance(result, Exception): |
| 236 | handle_error(result) |
| 237 | return f"{source} search failed: {str(result)}" |
| 238 | return result if result else "" |
| 239 | |
| 240 | def format_result_searxng(self, result, source): |
| 241 | if isinstance(result, Exception): |
| 242 | handle_error(result) |
| 243 | return f"{source} search failed: {str(result)}" |
| 244 | |
| 245 | if not result or "results" not in result: |
| 246 | return "" |
| 247 | |
| 248 | outputs = [] |
| 249 | for item in result["results"]: |
| 250 | if "qa" in item: |
| 251 | outputs.append( |
| 252 | f"## Next Result\n" |
| 253 | f"*Title*: {item['title'].strip()}\n" |
| 254 | f"*URL*: {item['url'].strip()}\n" |
| 255 | f"*Search Engine Summary*:\n{item['content'].strip()}\n" |
| 256 | f"*Query Result*:\n{item['qa'].strip()}" |
| 257 | ) |
| 258 | else: |
| 259 | outputs.append( |
| 260 | f"## Next Result\n" |
| 261 | f"*Title*: {item['title'].strip()}\n" |
| 262 | f"*URL*: {item['url'].strip()}\n" |
| 263 | f"*Search Engine Summary*:\n{item['content'].strip()}" |
| 264 | ) |
| 265 | |
| 266 | return "\n\n".join(outputs[:SEARCH_ENGINE_RESULTS]).strip() |