feat: extract document_query into _document_query plugin with parser strategy pattern
- Create plugins/_document_query/ with full plugin structure: plugin.yaml, default_config.yaml, tools/, helpers/, helpers/parsers/, prompts/, README.md - Add BaseParser ABC with asyncio.to_thread offload and configurable timeouts - Implement 5 parsers: PDF (PyMuPDF+Tesseract), HTML (Markdownify), Text (expanded mimetypes: YAML, XML, TOML, JS, TS, shell), Image (Unstructured), Unstructured (catch-all) - Add MIME type registry with priority-based routing via get_parser_for_mimetype() - Add gather_timeout on asyncio.gather for bounded concurrent fetches - All config externalized to default_config.yaml - Disable core files (._py.bak) replaced by plugin - Update knowledge_tool._py import to plugin path
Deimos Agent committed
Apr 17, 2026 at 13:17 UTC
5fd7a6a79e5a352b727ea1615a85a5f5bd7a151f
22 files changed
+752
-1
helpers/document_query._py.bak
renamed
plugins/_document_query/README.md
new
+42
@@ -0,0 +1,42 @@
1
+# Document Query Plugin
2
+
3
+Load, parse, index, and Q&A over local and remote documents with configurable
4
+timeouts and thread-safe parsers.
5
+
6
+## Features
7
+
8
+- **Strategy-pattern parsers** - MIME-type routing to dedicated parser classes
9
+- **Thread-safe execution** - all sync parsers offloaded to asyncio.to_thread
10
+- **Configurable timeouts** - per-document and gather-level timeouts
11
+- **Expanded format support** - PDF, HTML, text, YAML, XML, TOML, JS, TS, images, and catch-all Unstructured
12
+
13
+## Configuration
14
+
15
+See default_config.yaml for all options. Key settings:
16
+
17
+| Setting | Default | Description |
18
+|---------|---------|-------------|
19
+| fetch_timeout | 30 | HTTP fetch timeout (seconds) |
20
+| per_document_timeout | 60 | Max time for a single document parse |
21
+| gather_timeout | 120 | Max time for all documents combined |
22
+| chunk_size | 1000 | Text splitter chunk size |
23
+| chunk_overlap | 100 | Text splitter overlap |
24
+| search_threshold | 0.5 | Similarity search threshold |
25
+| thread_offload | true | Offload sync parsers to thread pool |
26
+
27
+## Parsers
28
+
29
+| Parser | MIME Types | Backend |
30
+|--------|-----------|---------|
31
+| PdfParser | application/pdf | PyMuPDF + Tesseract OCR fallback |
32
+| HtmlParser | text/html | Markdownify transformer |
33
+| TextParser | text/*, application/json, YAML, XML, TOML, JS, TS, shell | Direct read |
34
+| ImageParser | image/* | UnstructuredLoader |
35
+| UnstructuredParser | * (catch-all) | UnstructuredLoader hi-res |
36
+
37
+## Adding a new parser
38
+
39
+1. Create helpers/parsers/<format>.py extending BaseParser
40
+2. Set mimetypes class attribute
41
+3. Implement _parse_sync(document_uri, scheme)
42
+4. Register in helpers/parsers/__init__.py
plugins/_document_query/default_config.yaml
new
+18
@@ -0,0 +1,18 @@
1
+# Document Query Plugin Configuration
2
+# All timeout values in seconds
3
+
4
+# --- Timeouts ---
5
+fetch_timeout: 30 # HTTP fetch connect/read timeout
6
+per_document_timeout: 60 # max time for a single document parse
7
+gather_timeout: 120 # max time for all documents combined in one call
8
+
9
+# --- Parser settings ---
10
+chunk_size: 1000
11
+chunk_overlap: 100
12
+search_threshold: 0.5
13
+search_limit: 100
14
+max_remote_bytes: 52428800 # 50 MB
15
+
16
+# --- Feature flags ---
17
+pdf_ocr_fallback: true # enable Tesseract fallback for PDFs
18
+thread_offload: true # offload sync parsers to thread pool
plugins/_document_query/helpers/__init__.py
plugins/_document_query/helpers/document_query.py
new
+355
@@ -0,0 +1,355 @@
1
+"""Document query helper with thread-safe parsers and configurable timeouts.
2
+
3
+Extracted from the monolithic helpers/document_query.py into a plugin
4
+with parser strategy pattern where every document parser is offloaded to
5
+a thread pool and bounded by configurable timeouts.
6
+"""
7
+
8
+import asyncio
9
+import json
10
+import mimetypes
11
+import os
12
+from datetime import datetime
13
+from typing import Callable, List, Optional, Sequence, Tuple
14
+from urllib.parse import urlparse
15
+
16
+import aiohttp
17
+from langchain.schema import SystemMessage, HumanMessage
18
+from langchain.text_splitter import RecursiveCharacterTextSplitter
19
+from langchain_core.documents import Document
20
+
21
+from helpers import files, errors
22
+from helpers.print_style import PrintStyle
23
+from helpers.vector_db import VectorDB
24
+from agent import Agent
25
+
26
+from plugins._document_query.helpers.parsers import get_parser_for_mimetype
27
+
28
+
29
+DEFAULT_SEARCH_THRESHOLD = 0.5
30
+
31
+
32
+def _load_config(agent: Agent) -> dict:
33
+ """Load plugin config with fallback to defaults."""
34
+ from helpers.plugins import get_plugin_config
35
+ return get_plugin_config("_document_query", agent=agent) or {}
36
+
37
+
38
+class DocumentQueryStore:
39
+ """FAISS Store for document query results."""
40
+
41
+ DEFAULT_CHUNK_SIZE = 1000
42
+ DEFAULT_CHUNK_OVERLAP = 100
43
+
44
+ @staticmethod
45
+ def get(agent: Agent):
46
+ if not agent or not agent.config:
47
+ raise ValueError("Agent and agent config must be provided")
48
+ return DocumentQueryStore(agent)
49
+
50
+ def __init__(self, agent: Agent):
51
+ self.agent = agent
52
+ self.vector_db: VectorDB | None = None
53
+ self.config = _load_config(agent)
54
+
55
+ @staticmethod
56
+ def normalize_uri(uri: str) -> str:
57
+ normalized = uri.strip()
58
+ parsed = urlparse(normalized)
59
+ scheme = parsed.scheme or "file"
60
+ if scheme == "file":
61
+ path = files.fix_dev_path(
62
+ normalized.removeprefix("file://").removeprefix("file:")
63
+ )
64
+ normalized = f"file://{path}"
65
+ elif scheme in ["http", "https"]:
66
+ normalized = normalized.replace("http://", "https://")
67
+ return normalized
68
+
69
+ def init_vector_db(self):
70
+ return VectorDB(self.agent, cache=True)
71
+
72
+ async def add_document(
73
+ self, text: str, document_uri: str, metadata: dict | None = None
74
+ ) -> tuple[bool, list[str]]:
75
+ document_uri = self.normalize_uri(document_uri)
76
+ await self.delete_document(document_uri)
77
+ doc_metadata = metadata or {}
78
+ doc_metadata["document_uri"] = document_uri
79
+ doc_metadata["timestamp"] = datetime.now().strftime("%Y-%m-%d %H:%M:%S")
80
+ chunk_size = self.config.get("chunk_size", self.DEFAULT_CHUNK_SIZE)
81
+ chunk_overlap = self.config.get("chunk_overlap", self.DEFAULT_CHUNK_OVERLAP)
82
+ text_splitter = RecursiveCharacterTextSplitter(
83
+ chunk_size=chunk_size, chunk_overlap=chunk_overlap
84
+ )
85
+ chunks = text_splitter.split_text(text)
86
+ docs = []
87
+ for i, chunk in enumerate(chunks):
88
+ chunk_metadata = doc_metadata.copy()
89
+ chunk_metadata["chunk_index"] = i
90
+ chunk_metadata["total_chunks"] = len(chunks)
91
+ docs.append(Document(page_content=chunk, metadata=chunk_metadata))
92
+ if not docs:
93
+ PrintStyle.error(f"No chunks created for document: {document_uri}")
94
+ return False, []
95
+ try:
96
+ if not self.vector_db:
97
+ self.vector_db = self.init_vector_db()
98
+ ids = await self.vector_db.insert_documents(docs)
99
+ PrintStyle.standard(f"Added document '{document_uri}' with {len(docs)} chunks")
100
+ return True, ids
101
+ except Exception as e:
102
+ err_text = errors.format_error(e)
103
+ PrintStyle.error(f"Error adding document '{document_uri}': {err_text}")
104
+ return False, []
105
+
106
+ async def get_document(self, document_uri: str) -> Optional[Document]:
107
+ if not self.vector_db:
108
+ return None
109
+ document_uri = self.normalize_uri(document_uri)
110
+ docs = await self._get_document_chunks(document_uri)
111
+ if not docs:
112
+ return None
113
+ chunks = sorted(docs, key=lambda x: x.metadata.get("chunk_index", 0))
114
+ full_content = "\n".join(chunk.page_content for chunk in chunks)
115
+ metadata = chunks[0].metadata.copy()
116
+ metadata.pop("chunk_index", None)
117
+ metadata.pop("total_chunks", None)
118
+ return Document(page_content=full_content, metadata=metadata)
119
+
120
+ async def _get_document_chunks(self, document_uri: str) -> List[Document]:
121
+ if not self.vector_db:
122
+ return []
123
+ document_uri = self.normalize_uri(document_uri)
124
+ chunks = await self.vector_db.search_by_metadata(
125
+ filter=f"document_uri == '{document_uri}'",
126
+ )
127
+ PrintStyle.standard(f"Found {len(chunks)} chunks for document: {document_uri}")
128
+ return chunks
129
+
130
+ async def document_exists(self, document_uri: str) -> bool:
131
+ if not self.vector_db:
132
+ return False
133
+ document_uri = self.normalize_uri(document_uri)
134
+ chunks = await self._get_document_chunks(document_uri)
135
+ return len(chunks) > 0
136
+
137
+ async def delete_document(self, document_uri: str) -> bool:
138
+ if not self.vector_db:
139
+ return False
140
+ document_uri = self.normalize_uri(document_uri)
141
+ chunks = await self.vector_db.search_by_metadata(
142
+ filter=f"document_uri == '{document_uri}'",
143
+ )
144
+ if not chunks:
145
+ return False
146
+ ids_to_delete = [chunk.metadata["id"] for chunk in chunks]
147
+ if ids_to_delete:
148
+ dels = await self.vector_db.delete_documents_by_ids(ids_to_delete)
149
+ PrintStyle.standard(f"Deleted document '{document_uri}' with {len(dels)} chunks")
150
+ return True
151
+ return False
152
+
153
+ async def search_documents(
154
+ self, query: str, limit: int = 10, threshold: float = 0.5, filter: str = ""
155
+ ) -> List[Document]:
156
+ if not self.vector_db:
157
+ return []
158
+ if not query:
159
+ return []
160
+ try:
161
+ results = await self.vector_db.search_by_similarity_threshold(
162
+ query=query, limit=limit, threshold=threshold, filter=filter
163
+ )
164
+ PrintStyle.standard(f"Search '{query}' returned {len(results)} results")
165
+ return results
166
+ except Exception as e:
167
+ PrintStyle.error(f"Error searching documents: {str(e)}")
168
+ return []
169
+
170
+ async def search_document(
171
+ self, document_uri: str, query: str, limit: int = 10, threshold: float = 0.5
172
+ ) -> List[Document]:
173
+ return await self.search_documents(
174
+ query, limit, threshold, f"document_uri == '{document_uri}'"
175
+ )
176
+
177
+ async def list_documents(self) -> List[str]:
178
+ if not self.vector_db:
179
+ return []
180
+ uris = set()
181
+ for doc in self.vector_db.db.get_all_docs().values():
182
+ if isinstance(doc.metadata, dict):
183
+ uri = doc.metadata.get("document_uri")
184
+ if uri:
185
+ uris.add(uri)
186
+ return sorted(list(uris))
187
+
188
+
189
+class DocumentQueryHelper:
190
+
191
+ def __init__(
192
+ self, agent: Agent, progress_callback: Callable[[str], None] | None = None
193
+ ):
194
+ self.agent = agent
195
+ self.store = DocumentQueryStore.get(agent)
196
+ self.progress_callback = progress_callback or (lambda x: None)
197
+ self.store_lock = asyncio.Lock()
198
+ self.config = _load_config(agent)
199
+
200
+ async def document_qa(
201
+ self, document_uris: List[str], questions: Sequence[str]
202
+ ) -> Tuple[bool, str]:
203
+ self.progress_callback(f"Starting Q&A process for {len(document_uris)} documents")
204
+ await self.agent.handle_intervention()
205
+
206
+ gather_timeout = self.config.get("gather_timeout", 120)
207
+ try:
208
+ await asyncio.wait_for(
209
+ asyncio.gather(
210
+ *[self.document_get_content(uri, True) for uri in document_uris]
211
+ ),
212
+ timeout=gather_timeout,
213
+ )
214
+ except asyncio.TimeoutError:
215
+ raise ValueError(f"Document indexing timed out after {gather_timeout}s")
216
+
217
+ await self.agent.handle_intervention()
218
+ search_threshold = self.config.get("search_threshold", DEFAULT_SEARCH_THRESHOLD)
219
+ search_limit = self.config.get("search_limit", 100)
220
+ selected_chunks = {}
221
+
222
+ for question in questions:
223
+ self.progress_callback(f"Optimizing query: {question}")
224
+ await self.agent.handle_intervention()
225
+ system_content = self.agent.parse_prompt("fw.document_query.optmimize_query.md")
226
+ optimized_query = (
227
+ await self.agent.call_utility_model(
228
+ system=system_content, message=f'Search Query: "{question}"',
229
+ )
230
+ ).strip()
231
+
232
+ await self.agent.handle_intervention()
233
+ self.progress_callback(f"Searching documents with query: {optimized_query}")
234
+ normalized_uris = [self.store.normalize_uri(uri) for uri in document_uris]
235
+ doc_filter = " or ".join(
236
+ [f"document_uri == '{uri}'" for uri in normalized_uris]
237
+ )
238
+ chunks = await self.store.search_documents(
239
+ query=optimized_query, limit=search_limit,
240
+ threshold=search_threshold, filter=doc_filter,
241
+ )
242
+ self.progress_callback(f"Found {len(chunks)} chunks")
243
+ for chunk in chunks:
244
+ selected_chunks[chunk.metadata["id"]] = chunk
245
+
246
+ if not selected_chunks:
247
+ self.progress_callback("No relevant content found in the documents")
248
+ content = f"!!! No content found for documents: {json.dumps(document_uris)} matching queries: {json.dumps(questions)}"
249
+ return False, content
250
+
251
+ self.progress_callback(
252
+ f"Processing {len(questions)} questions in context of {len(selected_chunks)} chunks"
253
+ )
254
+ await self.agent.handle_intervention()
255
+ questions_str = "\n".join([f" * {question}" for question in questions])
256
+ content = "\n\n----\n\n".join(
257
+ [chunk.page_content for chunk in selected_chunks.values()]
258
+ )
259
+ qa_system_message = self.agent.parse_prompt("fw.document_query.system_prompt.md")
260
+ qa_user_message = f"# Document:\n{content}\n\n# Queries:\n{questions_str}"
261
+ ai_response, _reasoning = await self.agent.call_chat_model(
262
+ messages=[
263
+ SystemMessage(content=qa_system_message),
264
+ HumanMessage(content=qa_user_message),
265
+ ],
266
+ explicit_caching=False,
267
+ )
268
+ self.progress_callback(f"Q&A process completed")
269
+ return True, str(ai_response)
270
+
271
+ async def document_get_content(
272
+ self, document_uri: str, add_to_db: bool = False
273
+ ) -> str:
274
+ self.progress_callback(f"Fetching document content")
275
+ await self.agent.handle_intervention()
276
+ url = urlparse(document_uri)
277
+ scheme = url.scheme or "file"
278
+ mimetype, encoding = mimetypes.guess_type(document_uri)
279
+ mimetype = mimetype or "application/octet-stream"
280
+
281
+ if mimetype == "application/octet-stream":
282
+ if url.scheme in ["http", "https"]:
283
+ response = None
284
+ retries = 0
285
+ last_error = ""
286
+ while not response and retries < 3:
287
+ try:
288
+ async with aiohttp.ClientSession() as session:
289
+ response = await session.head(
290
+ document_uri,
291
+ timeout=aiohttp.ClientTimeout(total=2.0),
292
+ allow_redirects=True,
293
+ )
294
+ if response.status > 399:
295
+ raise Exception(response.status)
296
+ break
297
+ except Exception as e:
298
+ await asyncio.sleep(1)
299
+ last_error = str(e)
300
+ retries += 1
301
+ await self.agent.handle_intervention()
302
+ if not response:
303
+ raise ValueError(f"Document fetch error: {document_uri} ({last_error})")
304
+ mimetype = response.headers["content-type"]
305
+ if "content-length" in response.headers:
306
+ content_length = float(response.headers["content-length"]) / 1024 / 1024
307
+ if content_length > 50.0:
308
+ raise ValueError(f"Document exceeds max 50MB: {content_length} MB ({document_uri})")
309
+ if mimetype and "; charset=" in mimetype:
310
+ mimetype = mimetype.split("; charset=")[0]
311
+
312
+ if scheme == "file":
313
+ try:
314
+ document_uri = files.fix_dev_path(url.path)
315
+ except Exception as e:
316
+ raise ValueError(f"Invalid document path '{url.path}'") from e
317
+
318
+ if encoding:
319
+ raise ValueError(f"Compressed documents are unsupported '{encoding}' ({document_uri})")
320
+ if mimetype == "application/octet-stream":
321
+ raise ValueError(f"Unsupported document mimetype '{mimetype}' ({document_uri})")
322
+
323
+ document_uri_norm = self.store.normalize_uri(document_uri)
324
+ await self.agent.handle_intervention()
325
+ exists = await self.store.document_exists(document_uri_norm)
326
+ document_content = ""
327
+
328
+ if not exists:
329
+ await self.agent.handle_intervention()
330
+ parser = get_parser_for_mimetype(mimetype)
331
+ if parser is None:
332
+ raise ValueError(f"No parser found for mimetype '{mimetype}' ({document_uri})")
333
+ per_doc_timeout = self.config.get("per_document_timeout", 60)
334
+ thread_offload = self.config.get("thread_offload", True)
335
+ document_content = await parser.parse(
336
+ document_uri=document_uri, scheme=scheme,
337
+ timeout=per_doc_timeout, thread_offload=thread_offload,
338
+ )
339
+ if add_to_db:
340
+ self.progress_callback(f"Indexing document")
341
+ await self.agent.handle_intervention()
342
+ async with self.store_lock:
343
+ success, ids = await self.store.add_document(document_content, document_uri_norm)
344
+ if not success:
345
+ self.progress_callback(f"Failed to index document")
346
+ raise ValueError(f"Failed to index document: {document_uri_norm}")
347
+ self.progress_callback(f"Indexed {len(ids)} chunks")
348
+ else:
349
+ await self.agent.handle_intervention()
350
+ doc = await self.store.get_document(document_uri_norm)
351
+ if doc:
352
+ document_content = doc.page_content
353
+ else:
354
+ raise ValueError(f"Document not found: {document_uri_norm}")
355
+ return document_content
plugins/_document_query/helpers/parsers/__init__.py
new
+17
@@ -0,0 +1,17 @@
1
+"""Document parser registry."""
2
+from .base import BaseParser
3
+from .pdf import PdfParser
4
+from .html import HtmlParser
5
+from .text import TextParser
6
+from .image import ImageParser
7
+from .unstructured import UnstructuredParser
8
+
9
+_PARSERS = [PdfParser(), HtmlParser(), TextParser(), ImageParser(), UnstructuredParser()]
10
+
11
+def get_parser_for_mimetype(mimetype: str) -> BaseParser | None:
12
+ for parser in _PARSERS:
13
+ if parser.can_handle(mimetype):
14
+ return parser
15
+ return None
16
+
17
+__all__ = ["BaseParser", "PdfParser", "HtmlParser", "TextParser", "ImageParser", "UnstructuredParser", "get_parser_for_mimetype"]
plugins/_document_query/helpers/parsers/base.py
new
+62
@@ -0,0 +1,62 @@
1
+"""Base parser with built-in thread offload and timeout."""
2
+
3
+import asyncio
4
+from abc import ABC, abstractmethod
5
+from typing import Optional
6
+
7
+from helpers.print_style import PrintStyle
8
+
9
+
10
+class BaseParser(ABC):
11
+ """Abstract base for document parsers.
12
+
13
+ Every parser runs synchronously but is automatically offloaded to a
14
+ thread pool and bounded by a configurable timeout when called through
15
+ parse(). This prevents any single parser from blocking the asyncio
16
+ event loop.
17
+ """
18
+
19
+ mimetypes: list[str] = []
20
+
21
+ def can_handle(self, mimetype: str) -> bool:
22
+ """Return True if this parser supports *mimetype*."""
23
+ for pattern in self.mimetypes:
24
+ if pattern == "*":
25
+ return True
26
+ if pattern.endswith("/"):
27
+ if mimetype.startswith(pattern):
28
+ return True
29
+ elif mimetype == pattern:
30
+ return True
31
+ return False
32
+
33
+ async def parse(
34
+ self,
35
+ document_uri: str,
36
+ scheme: str,
37
+ timeout: float = 60.0,
38
+ thread_offload: bool = True,
39
+ ) -> str:
40
+ try:
41
+ if thread_offload:
42
+ return await asyncio.wait_for(
43
+ asyncio.to_thread(self._parse_sync, document_uri, scheme),
44
+ timeout=timeout,
45
+ )
46
+ else:
47
+ return await asyncio.wait_for(
48
+ self._parse_async(document_uri, scheme),
49
+ timeout=timeout,
50
+ )
51
+ except asyncio.TimeoutError:
52
+ PrintStyle.error(
53
+ f"Parser {self.__class__.__name__} timed out after {timeout}s on {document_uri}"
54
+ )
55
+ raise ValueError(f"Document parsing timed out after {timeout}s: {document_uri}")
56
+
57
+ async def _parse_async(self, document_uri: str, scheme: str) -> str:
58
+ return self._parse_sync(document_uri, scheme)
59
+
60
+ @abstractmethod
61
+ def _parse_sync(self, document_uri: str, scheme: str) -> str:
62
+ ...
plugins/_document_query/helpers/parsers/html.py
new
+22
@@ -0,0 +1,22 @@
1
+"""HTML parser using Markdownify transformer."""
2
+
3
+from langchain_community.document_loaders import AsyncHtmlLoader
4
+from langchain_community.document_transformers import MarkdownifyTransformer
5
+from langchain_core.documents import Document
6
+
7
+from helpers import files
8
+from .base import BaseParser
9
+
10
+
11
+class HtmlParser(BaseParser):
12
+ mimetypes = ["text/html"]
13
+
14
+ def _parse_sync(self, document_uri: str, scheme: str) -> str:
15
+ if scheme in ["http", "https"]:
16
+ parts = AsyncHtmlLoader(web_path=document_uri).load()
17
+ elif scheme == "file":
18
+ content = files.read_file_bin(document_uri).decode("utf-8")
19
+ parts = [Document(page_content=content, metadata={"source": document_uri})]
20
+ else:
21
+ raise ValueError(f"Unsupported scheme: {scheme}")
22
+ return "\n".join(e.page_content for e in MarkdownifyTransformer().transform_documents(parts))
plugins/_document_query/helpers/parsers/image.py
new
+9
@@ -0,0 +1,9 @@
1
+"""Image parser — delegates to UnstructuredLoader."""
2
+from .base import BaseParser
3
+from .unstructured import UnstructuredParser
4
+
5
+
6
+class ImageParser(BaseParser):
7
+ mimetypes = ["image/"]
8
+ def _parse_sync(self, document_uri: str, scheme: str) -> str:
9
+ return UnstructuredParser()._parse_sync(document_uri, scheme)
plugins/_document_query/helpers/parsers/pdf.py
new
+63
@@ -0,0 +1,63 @@
1
+"""PDF parser with PyMuPDF primary and Tesseract OCR fallback."""
2
+
3
+import os
4
+import tempfile
5
+
6
+from langchain_community.document_loaders.pdf import PyMuPDFLoader
7
+from langchain_community.document_loaders.parsers.images import TesseractBlobParser
8
+from langchain_core.documents import Document
9
+
10
+from helpers import files
11
+from helpers.print_style import PrintStyle
12
+
13
+from .base import BaseParser
14
+
15
+
16
+class PdfParser(BaseParser):
17
+ mimetypes = ["application/pdf"]
18
+
19
+ def _parse_sync(self, document_uri: str, scheme: str) -> str:
20
+ temp_file_path = ""
21
+ if scheme == "file":
22
+ file_content_bytes = files.read_file_bin(document_uri)
23
+ with tempfile.NamedTemporaryFile(delete=False, suffix=".pdf") as f:
24
+ f.write(file_content_bytes)
25
+ temp_file_path = f.name
26
+ elif scheme in ["http", "https"]:
27
+ import requests
28
+ with tempfile.NamedTemporaryFile(delete=False, suffix=".pdf") as f:
29
+ resp = requests.get(document_uri, timeout=10.0)
30
+ if resp.status_code != 200:
31
+ raise ValueError(f"Failed to download PDF: {resp.status_code}")
32
+ f.write(resp.content)
33
+ temp_file_path = f.name
34
+ else:
35
+ raise ValueError(f"Unsupported scheme: {scheme}")
36
+
37
+ if not os.path.exists(temp_file_path):
38
+ raise ValueError(f"Temporary file not found: {temp_file_path}")
39
+ try:
40
+ contents = self._parse_with_pymupdf(temp_file_path)
41
+ if not contents:
42
+ contents = self._parse_with_ocr(temp_file_path)
43
+ return contents
44
+ finally:
45
+ os.unlink(temp_file_path)
46
+
47
+ def _parse_with_pymupdf(self, file_path: str) -> str:
48
+ try:
49
+ loader = PyMuPDFLoader(
50
+ file_path, mode="single", extract_tables="markdown",
51
+ extract_images=True, images_inner_format="text",
52
+ images_parser=TesseractBlobParser(), pages_delimiter="\n",
53
+ )
54
+ return "\n".join(e.page_content for e in loader.load())
55
+ except Exception as e:
56
+ PrintStyle.error(f"PyMuPDF parsing failed: {e}")
57
+ return ""
58
+
59
+ def _parse_with_ocr(self, file_path: str) -> str:
60
+ import pdf2image, pytesseract
61
+ PrintStyle.debug(f"FALLBACK: OCR for {file_path}")
62
+ pages = pdf2image.convert_from_path(file_path)
63
+ return "\n\n".join(pytesseract.image_to_string(p) for p in pages)
plugins/_document_query/helpers/parsers/text.py
new
+26
@@ -0,0 +1,26 @@
1
+"""Plain text and config document parser."""
2
+
3
+from langchain_community.document_loaders import AsyncHtmlLoader
4
+from langchain_core.documents import Document
5
+
6
+from helpers import files
7
+from .base import BaseParser
8
+
9
+
10
+class TextParser(BaseParser):
11
+ mimetypes = [
12
+ "text/", "application/json", "application/yaml", "application/x-yaml",
13
+ "application/xml", "application/toml", "application/x-toml",
14
+ "application/javascript", "application/typescript",
15
+ "application/x-sh", "application/x-shellscript",
16
+ ]
17
+
18
+ def _parse_sync(self, document_uri: str, scheme: str) -> str:
19
+ if scheme in ["http", "https"]:
20
+ elements = AsyncHtmlLoader(web_path=document_uri).load()
21
+ elif scheme == "file":
22
+ content = files.read_file_bin(document_uri).decode("utf-8")
23
+ elements = [Document(page_content=content, metadata={"source": document_uri})]
24
+ else:
25
+ raise ValueError(f"Unsupported scheme: {scheme}")
26
+ return "\n".join(e.page_content for e in elements)
plugins/_document_query/helpers/parsers/unstructured.py
new
+33
@@ -0,0 +1,33 @@
1
+"""Catch-all parser using UnstructuredLoader."""
2
+
3
+import os
4
+import tempfile
5
+
6
+from helpers import files
7
+from .base import BaseParser
8
+
9
+os.environ.setdefault("USER_AGENT", "@mixedbread-ai/unstructured")
10
+from langchain_unstructured import UnstructuredLoader
11
+
12
+
13
+class UnstructuredParser(BaseParser):
14
+ mimetypes = ["*"]
15
+
16
+ def _parse_sync(self, document_uri: str, scheme: str) -> str:
17
+ if scheme in ["http", "https"]:
18
+ loader = UnstructuredLoader(web_url=document_uri, mode="single", partition_via_api=False, strategy="hi_res")
19
+ elements = loader.load()
20
+ elif scheme == "file":
21
+ content = files.read_file_bin(document_uri)
22
+ _, ext = os.path.splitext(document_uri)
23
+ with tempfile.NamedTemporaryFile(delete=False, suffix=ext) as f:
24
+ f.write(content)
25
+ tmp = f.name
26
+ try:
27
+ loader = UnstructuredLoader(file_path=tmp, mode="single", partition_via_api=False, strategy="hi_res")
28
+ elements = loader.load()
29
+ finally:
30
+ if os.path.exists(tmp): os.unlink(tmp)
31
+ else:
32
+ raise ValueError(f"Unsupported scheme: {scheme}")
33
+ return "\n".join(e.page_content for e in elements)
plugins/_document_query/plugin.yaml
new
+8
@@ -0,0 +1,8 @@
1
+name: _document_query
2
+title: Document Query
3
+description: Load, parse, index, and Q&A over local and remote documents with configurable timeouts and thread-safe parsers.
4
+version: 1.0.0
5
+settings_sections:
6
+ - agent
7
+per_project_config: true
8
+per_agent_config: false
plugins/_document_query/prompts/agent.system.tool.document_query.md
new
+8
@@ -0,0 +1,8 @@
1
+### document_query
2
+read local or remote documents or answer questions about them
3
+args:
4
+- `document`: url path or list of them
5
+- `queries`: optional list of questions
6
+- `query`: optional single-question alias
7
+without `query` or `queries` it returns document content
8
+for local files use full path; for web documents use full urls
plugins/_document_query/prompts/fw.document_query.optmimize_query.md
renamed
plugins/_document_query/prompts/fw.document_query.system_prompt.md
renamed
plugins/_document_query/tools/document_query.py
new
+55
@@ -0,0 +1,55 @@
1
+import asyncio
2
+
3
+from helpers.tool import Tool, Response
4
+from plugins._document_query.helpers.document_query import DocumentQueryHelper
5
+
6
+
7
+class DocumentQueryTool(Tool):
8
+
9
+ async def execute(self, **kwargs):
10
+ document_uri = kwargs.get("document")
11
+ document_uris = []
12
+
13
+ if isinstance(document_uri, list):
14
+ document_uris = document_uri
15
+ elif isinstance(document_uri, str):
16
+ document_uris = [document_uri]
17
+
18
+ if not document_uris:
19
+ return Response(message="Error: no document provided", break_loop=False)
20
+
21
+ queries = (
22
+ kwargs["queries"]
23
+ if "queries" in kwargs
24
+ else [kwargs["query"]]
25
+ if ("query" in kwargs and kwargs["query"])
26
+ else []
27
+ )
28
+ try:
29
+ progress = []
30
+
31
+ def progress_callback(msg):
32
+ progress.append(msg)
33
+ self.log.update(progress="\n".join(progress))
34
+
35
+ helper = DocumentQueryHelper(self.agent, progress_callback)
36
+ if not queries:
37
+ gather_timeout = helper.config.get("gather_timeout", 120)
38
+ try:
39
+ contents = await asyncio.wait_for(
40
+ asyncio.gather(
41
+ *[helper.document_get_content(uri) for uri in document_uris]
42
+ ),
43
+ timeout=gather_timeout,
44
+ )
45
+ except asyncio.TimeoutError:
46
+ return Response(
47
+ message=f"Error: document processing timed out after {gather_timeout}s",
48
+ break_loop=False,
49
+ )
50
+ content = "\n\n---\n\n".join(contents)
51
+ else:
52
+ _, content = await helper.document_qa(document_uris, queries)
53
+ return Response(message=content, break_loop=False)
54
+ except Exception as e: # pylint: disable=broad-exception-caught
55
+ return Response(message=f"Error processing document: {e}", break_loop=False)
prompts/agent.system.tool.document_query._md.bak
renamed
prompts/fw.document_query.optmimize_query._md.bak
new
+28
@@ -0,0 +1,28 @@
1
+# AI role
2
+- You are an AI assistant being part of a larger RAG system based on vector similarity search
3
+- Your job is to take a human written question and convert it into a concise vector store search query
4
+- The goal is to yield as many correct results and as few false positives as possible
5
+
6
+# Input
7
+- you are provided with original search query as user message
8
+
9
+# Response rules !!!
10
+- respond only with optimized result query text
11
+- no text before or after
12
+- no conversation, you are a tool agent, not a conversational agent
13
+
14
+# Optimized query
15
+- optimized query is consise, short and to the point
16
+- contains only keywords and phrases, no full sentences
17
+- include alternatives and variations for better coverage
18
+
19
+
20
+# Examples
21
+User: What is the capital of France?
22
+Agent: france capital city
23
+
24
+User: What does it say about transmission?
25
+Agent: transmission gearbox automatic manual
26
+
27
+User: What did John ask Monica on Tuesday?
28
+Agent: john monica conversation dialogue question ask tuesday
prompts/fw.document_query.system_prompt._md.bak
new
+5
@@ -0,0 +1,5 @@
1
+You are an AI assistant who can answer questions about a given document text.
2
+The assistant is part of a larger application that is used to answer questions about a document.
3
+The assistant is given a document and a list of queries and the assistant must answer the quries based on the document.
4
+!! The response should be in markdown format.
5
+!! The response should only include the queries as headings and the answers to the queries. The markdown should contain paragraphs with "#### <Query>" as headings (<Query> being the original query) followed by the query answer as the paragraph text content.
tools/document_query._py.bak
renamed
tools/knowledge_tool._py
+1
-1
@@ -4,7 +4,7 @@ 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 helpers.document_query import DocumentQueryHelper
7
+from plugins._document_query.helpers.document_query import DocumentQueryHelper
8
9
SEARCH_ENGINE_RESULTS = 10
10