(self, query: str, history: list)
| 219 | return "rag_query" |
| 220 | |
| 221 | def _run_graph_query(self, query: str, history: list) -> Dict[str, Any]: |
| 222 | contextual_query = self._format_query_with_history(query, history) |
| 223 | structured_query = self.graph_query_translator.translate(contextual_query) |
| 224 | if not structured_query.get("start_node"): |
| 225 | return self.retrieval_pipeline.run(contextual_query, window_size_override=0) |
| 226 | results = self.graph_retriever.retrieve(structured_query) |
| 227 | if not results: |
| 228 | return self.retrieval_pipeline.run(contextual_query, window_size_override=0) |
| 229 | answer = ", ".join([res['details']['node_id'] for res in results]) |
| 230 | return {"answer": f"From the knowledge graph: {answer}", "source_documents": results} |
| 231 | |
| 232 | def _get_cache_key(self, query: str, query_type: str) -> str: |
| 233 | """Generate a cache key for the query""" |
no test coverage detected