(
user_id: str,
question: str,
limit: int,
pinned_nodes: Optional[List[Dict[str, Any]]],
allowed_node_ids: Optional[Iterable[str]] = None,
)
| 135 | |
| 136 | |
| 137 | def _prepare_hits( |
| 138 | user_id: str, |
| 139 | question: str, |
| 140 | limit: int, |
| 141 | pinned_nodes: Optional[List[Dict[str, Any]]], |
| 142 | allowed_node_ids: Optional[Iterable[str]] = None, |
| 143 | ) -> tuple[List[Dict[str, Any]], Optional[str]]: |
| 144 | embedding_error = None |
| 145 | try: |
| 146 | wiki_db.embed_nodes_for_user(user_id, limit=500) |
| 147 | except Exception as exc: |
| 148 | embedding_error = str(exc) |
| 149 | retrieved = wiki_db.search_nodes(user_id, question, limit=max(limit, limit + len(pinned_nodes or []))) |
| 150 | if allowed_node_ids is not None: |
| 151 | allowed = {str(node_id) for node_id in allowed_node_ids if str(node_id or "").strip()} |
| 152 | retrieved = [node for node in retrieved or [] if str(node.get("node_id") or "") in allowed] |
| 153 | return _merge_hits(pinned_nodes or [], retrieved, limit), embedding_error |
| 154 | |
| 155 | |
| 156 | def _build_citations(user_id: str, hits: List[Dict[str, Any]]) -> List[Dict[str, Any]]: |
no test coverage detected