Coverage for src/local_deep_research/citation_handlers/standard_citation_handler.py: 100%
49 statements
« prev ^ index » next coverage.py v7.16.2, created at 2026-10-22 21:20 +0000
1"""
2Standard citation handler - the original implementation.
3"""
5from datetime import datetime, timezone
6from typing import Any, Dict, List, Union
8from langchain_core.documents import Document
10from ..security.log_sanitizer import sanitize_error_message, strip_control_chars
11from .base_citation_handler import BaseCitationHandler
14def _identity_text(value: Any) -> str:
15 """Render scalar metadata; containers could hide secrets from the scrubber."""
16 if value is None:
17 return ""
18 if isinstance(value, (str, int, float)):
19 return str(value)
20 return "[non-text value omitted]"
23class StandardCitationHandler(BaseCitationHandler):
24 """Standard citation handler with detailed analysis."""
26 def _format_sources(self, documents: List[Document]) -> str:
27 """Keep source identities alongside the text used for synthesis."""
28 sources = []
29 for doc in documents:
30 # Bound the title and URL and keep each on one line so they cannot
31 # add identity lines; page_content passes through unchanged, as in
32 # the base formatter. Preserve the original metadata.
33 title = strip_control_chars(
34 _identity_text(doc.metadata.get("title"))
35 )
36 title = sanitize_error_message(" ".join(title.split()))[:200]
37 # Unlike the log-sink scrubber, this preserves query parameters
38 # that identify sources (e.g. a video's ?v=), redacting credentials.
39 url = strip_control_chars(
40 _identity_text(doc.metadata.get("source"))
41 )
42 url = sanitize_error_message(" ".join(url.split()))
43 # Scrub the whole URL first so userinfo containing "#" is caught,
44 # then apply the query-parameter scrubber to fragment parameters.
45 url, separator, fragment = url.partition("#")
46 if separator:
47 url += "#" + sanitize_error_message("?" + fragment)[1:]
48 # Sanitize before truncating so a cut cannot hide a credential's
49 # identifying suffix (e.g. the @ in URL userinfo).
50 url = url[:2000]
51 sources.append(
52 f"[{doc.metadata['index']}] Title: {title}\n"
53 f"URL: {url}\nContent: {doc.page_content}"
54 )
55 return "\n\n".join(sources)
57 def analyze_initial(
58 self, query: str, search_results: Union[str, List[Dict]]
59 ) -> Dict[str, Any]:
60 documents = self._create_documents(search_results)
61 if not documents:
62 return self._no_sources_response(query)
63 formatted_sources = self._format_sources(documents)
64 current_timestamp = datetime.now(timezone.utc).strftime(
65 "%Y-%m-%d %H:%M"
66 )
68 output_prefix = self._get_output_instruction_prefix()
70 prompt = f"""{output_prefix}Analyze the following information concerning the question and include citations using numbers in square brackets [1], [2], etc. When citing, use the source number provided at the start of each source.
71The Title and URL lines identify each source for your reasoning; cite by number only — links are added automatically.
73Question: {query}
75Sources:
76{formatted_sources}
78Current time is {current_timestamp} UTC for verifying temporal references in sources.
80Provide a detailed analysis with inline citations. Citations must only appear inline within your sentences (e.g. "According to the research [1], ..."). Do NOT create a bibliography, sources list, or list of citations at the end, as sources will be provided automatically. Never make up sources. Only write text relevant to the question.
81"""
83 response = self._invoke_with_streaming(prompt)
84 return {"content": response, "documents": documents}
86 def analyze_followup(
87 self,
88 question: str,
89 search_results: Union[str, List[Dict]],
90 previous_knowledge: str,
91 nr_of_links: int,
92 ) -> Dict[str, Any]:
93 """Process follow-up analysis with citations."""
94 documents = self._create_documents(
95 search_results, nr_of_links=nr_of_links
96 )
97 # With previous knowledge present the LLM can still cite prior
98 # sources legitimately; with neither, refuse to synthesize.
99 if not documents and not (previous_knowledge or "").strip():
100 return self._no_sources_response(question)
101 formatted_sources = self._format_sources(documents)
102 # Add fact-checking step
103 fact_check_prompt = f"""Analyze these sources for factual consistency:
104The Title and URL lines identify each source for your reasoning; cite by number only — links are added automatically.
1061. Cross-reference major claims between sources
1072. Identify and flag any contradictions
1083. Verify basic facts (dates, company names, ownership)
1094. Note when sources disagree
111Previous Knowledge:
112{previous_knowledge}
114New Sources:
115{formatted_sources}
117 Return any inconsistencies or conflicts found."""
118 if self.is_fact_checking_enabled():
119 fact_check_response = self._invoke_text(fact_check_prompt)
120 else:
121 fact_check_response = ""
123 current_timestamp = datetime.now(timezone.utc).strftime(
124 "%Y-%m-%d %H:%M"
125 )
127 output_prefix = self._get_output_instruction_prefix()
129 prompt = f"""{output_prefix}Using the previous knowledge and new sources, answer the question. Include citations using numbers in square brackets [1], [2], etc. When citing, use the source number provided at the start of each source. Reflect information from sources critically.
130The Title and URL lines identify each source for your reasoning; cite by number only — links are added automatically.
132Previous Knowledge:
133{previous_knowledge}
135Question: {question}
137New Sources:
138{formatted_sources}
140Current time is {current_timestamp} UTC for verifying temporal references in sources.
142Reflect information from sources critically based on: {fact_check_response}. Never invent sources.
143Provide a detailed answer with inline citations. Citations must only appear inline within your sentences (e.g. "According to [1], ..."). Do NOT create a bibliography, sources list, or list of citations at the end, as sources will be provided automatically."""
145 response = self._invoke_with_streaming(prompt)
147 return {"content": response, "documents": documents}