query
Generate completion using relevant chunks as context.
def query(
query: str,
filters: Optional[Dict[str, Any]] = None,
k: int = 4,
min_score: float = 0.0,
max_tokens: Optional[int] = None,
temperature: Optional[float] = None,
use_colpali: bool = True,
graph_name: Optional[str] = None,
hop_depth: int = 1,
include_paths: bool = False,
prompt_overrides: Optional[Union[QueryPromptOverrides, Dict[str, Any]]] = None,
schema: Optional[Union[Type[BaseModel], Dict[str, Any]]] = None,
) -> CompletionResponse
def query(
query: str,
filters: Optional[Dict[str, Any]] = None,
k: int = 4,
min_score: float = 0.0,
max_tokens: Optional[int] = None,
temperature: Optional[float] = None,
use_colpali: bool = True,
graph_name: Optional[str] = None,
hop_depth: int = 1,
include_paths: bool = False,
prompt_overrides: Optional[Union[QueryPromptOverrides, Dict[str, Any]]] = None,
schema: Optional[Union[Type[BaseModel], Dict[str, Any]]] = None,
) -> CompletionResponse
async def query(
query: str,
filters: Optional[Dict[str, Any]] = None,
k: int = 4,
min_score: float = 0.0,
max_tokens: Optional[int] = None,
temperature: Optional[float] = None,
use_colpali: bool = True,
graph_name: Optional[str] = None,
hop_depth: int = 1,
include_paths: bool = False,
prompt_overrides: Optional[Union[QueryPromptOverrides, Dict[str, Any]]] = None,
schema: Optional[Union[Type[BaseModel], Dict[str, Any]]] = None,
) -> CompletionResponse
Parameters
query
(str): Query text
filters
(Dict[str, Any], optional): Optional metadata filters
k
(int, optional): Number of chunks to use as context. Defaults to 4.
min_score
(float, optional): Minimum similarity threshold. Defaults to 0.0.
max_tokens
(int, optional): Maximum tokens in completion
temperature
(float, optional): Model temperature
use_colpali
(bool, optional): Whether to use ColPali-style embedding model to generate the completion (only works for documents ingested with use_colpali=True
). Defaults to True.
graph_name
(str, optional): Optional name of the graph to use for knowledge graph-enhanced retrieval
hop_depth
(int, optional): Number of relationship hops to traverse in the graph (1-3). Defaults to 1.
include_paths
(bool, optional): Whether to include relationship paths in the response. Defaults to False.
prompt_overrides
(QueryPromptOverrides | Dict[str, Any], optional): Optional customizations for entity extraction, resolution, and query prompts
schema
(Type[BaseModel] | Dict[str, Any], optional): Optional schema for structured output, can be a Pydantic model or a JSON schema dict
Returns
CompletionResponse
: Response containing the completion, source information, and potentially structured output.
Examples
Standard Query
from morphik import Morphik
db = Morphik()
response = db.query(
"What are the key findings about customer satisfaction?",
filters={"department": "research"},
temperature=0.7
)
print(response.completion)
# Print the sources used for the completion
for source in response.sources:
print(f"Document ID: {source.document_id}, Chunk: {source.chunk_number}, Score: {source.score}")
from morphik import Morphik
db = Morphik()
response = db.query(
"What are the key findings about customer satisfaction?",
filters={"department": "research"},
temperature=0.7
)
print(response.completion)
# Print the sources used for the completion
for source in response.sources:
print(f"Document ID: {source.document_id}, Chunk: {source.chunk_number}, Score: {source.score}")
from morphik import AsyncMorphik
async with AsyncMorphik() as db:
response = await db.query(
"What are the key findings about customer satisfaction?",
filters={"department": "research"},
temperature=0.7
)
print(response.completion)
# Print the sources used for the completion
for source in response.sources:
print(f"Document ID: {source.document_id}, Chunk: {source.chunk_number}, Score: {source.score}")
Knowledge Graph Enhanced Query
from morphik import Morphik
db = Morphik()
# Use a knowledge graph to enhance the query
response = db.query(
"How does product X relate to customer segment Y?",
graph_name="market_graph",
hop_depth=2,
include_paths=True
)
print(response.completion)
# If include_paths=True, you can inspect the graph paths
if response.metadata and "graph" in response.metadata:
for path in response.metadata["graph"]["paths"]:
print(" -> ".join(path))
from morphik import Morphik
db = Morphik()
# Use a knowledge graph to enhance the query
response = db.query(
"How does product X relate to customer segment Y?",
graph_name="market_graph",
hop_depth=2,
include_paths=True
)
print(response.completion)
# If include_paths=True, you can inspect the graph paths
if response.metadata and "graph" in response.metadata:
for path in response.metadata["graph"]["paths"]:
print(" -> ".join(path))
from morphik import AsyncMorphik
async with AsyncMorphik() as db:
# Knowledge graph enhanced query
kg_response = await db.query(
"How does product X relate to customer segment Y?",
graph_name="market_graph",
hop_depth=2,
include_paths=True
)
print(kg_response.completion)
# If include_paths=True, you can inspect the graph paths
if kg_response.metadata and "graph" in kg_response.metadata:
for path in kg_response.metadata["graph"]["paths"]:
print(" -> ".join(path))
With Custom Prompt Overrides
from morphik import Morphik
from morphik.models import QueryPromptOverride, QueryPromptOverrides
db = Morphik()
# Using the QueryPromptOverrides object
response = db.query(
"What are the key findings?",
filters={"category": "research"},
prompt_overrides=QueryPromptOverrides(
query=QueryPromptOverride(
prompt_template="Answer the question in a formal, academic tone: {question}\n\nContext:\n{context}\n\nAnswer:"
)
)
)
# Alternatively, using a dictionary
response = db.query(
"What are the key findings?",
filters={"category": "research"},
prompt_overrides={
"query": {
"prompt_template": "Answer the question in a formal, academic tone: {question}\n\nContext:\n{context}\n\nAnswer:"
}
}
)
print(response.completion)
from morphik import Morphik
from morphik.models import QueryPromptOverride, QueryPromptOverrides
db = Morphik()
# Using the QueryPromptOverrides object
response = db.query(
"What are the key findings?",
filters={"category": "research"},
prompt_overrides=QueryPromptOverrides(
query=QueryPromptOverride(
prompt_template="Answer the question in a formal, academic tone: {question}\n\nContext:\n{context}\n\nAnswer:"
)
)
)
# Alternatively, using a dictionary
response = db.query(
"What are the key findings?",
filters={"category": "research"},
prompt_overrides={
"query": {
"prompt_template": "Answer the question in a formal, academic tone: {question}\n\nContext:\n{context}\n\nAnswer:"
}
}
)
print(response.completion)
from morphik import AsyncMorphik
from morphik.models import QueryPromptOverride, EntityExtractionPromptOverride, QueryPromptOverrides
async with AsyncMorphik() as db:
# Example with both query and entity extraction customization
response = await db.query(
"How does the medication affect diabetes?",
graph_name="medical_graph",
prompt_overrides=QueryPromptOverrides(
# Customize how responses are generated
query=QueryPromptOverride(
prompt_template="Provide a concise, medically accurate answer: {question}\n\nContext:\n{context}\n\nAnswer:"
),
# Customize entity extraction
entity_extraction=EntityExtractionPromptOverride(
examples=[
{"label": "Insulin", "type": "MEDICATION"},
{"label": "Diabetes", "type": "CONDITION"}
]
)
)
)
print(response.completion)
CompletionResponse Properties
The CompletionResponse
object returned by this method has the following properties:
completion
(str | Dict[str, Any] | None): The generated completion text or the structured output dictionary.
usage
(Dict[str, int]): Token usage information
sources
(List[ChunkSource]): Sources of chunks used in the completion
metadata
(Dict[str, Any], optional): Additional metadata about the completion. When using a knowledge graph with include_paths=True
, this contains graph traversal information.
finish_reason
(Optional[str]): Reason the generation finished (e.g., ‘stop’, ‘length’)
ChunkSource Properties
Each ChunkSource
object in the sources
list has the following properties:
document_id
(str): ID of the source document
chunk_number
(int): Chunk number within the document
score
(Optional[float]): Relevance score (if available)
Using Structured Output
Use the schema
parameter to get the completion response in a structured format according to a Pydantic model or a JSON schema dictionary.
from morphik import Morphik
from pydantic import BaseModel
from typing import List
# Define the desired output structure
class ResearchFindings(BaseModel):
main_finding: str
supporting_evidence: List[str]
limitations: List[str]
db = Morphik()
response = db.query(
"Summarize the key research findings from these documents",
filters={"department": "research"},
schema=ResearchFindings
)
# Check if the completion is a dictionary (structured output)
if isinstance(response.completion, dict):
try:
# Parse the dictionary into the Pydantic model
findings = ResearchFindings(**response.completion)
print(f"Main finding: {findings.main_finding}")
print("Supporting evidence:")
for evidence in findings.supporting_evidence:
print(f"- {evidence}")
print("Limitations:")
for limitation in findings.limitations:
print(f"- {limitation}")
except Exception as e:
print(f"Error parsing structured output: {e}")
# Fallback: print the raw dictionary
print(response.completion)
elif isinstance(response.completion, str):
# Fallback to text completion
print(response.completion)
from morphik import Morphik
from pydantic import BaseModel
from typing import List
# Define the desired output structure
class ResearchFindings(BaseModel):
main_finding: str
supporting_evidence: List[str]
limitations: List[str]
db = Morphik()
response = db.query(
"Summarize the key research findings from these documents",
filters={"department": "research"},
schema=ResearchFindings
)
# Check if the completion is a dictionary (structured output)
if isinstance(response.completion, dict):
try:
# Parse the dictionary into the Pydantic model
findings = ResearchFindings(**response.completion)
print(f"Main finding: {findings.main_finding}")
print("Supporting evidence:")
for evidence in findings.supporting_evidence:
print(f"- {evidence}")
print("Limitations:")
for limitation in findings.limitations:
print(f"- {limitation}")
except Exception as e:
print(f"Error parsing structured output: {e}")
# Fallback: print the raw dictionary
print(response.completion)
elif isinstance(response.completion, str):
# Fallback to text completion
print(response.completion)
from morphik import AsyncMorphik
from pydantic import BaseModel
from typing import List
# Define the desired output structure
class ResearchFindings(BaseModel):
main_finding: str
supporting_evidence: List[str]
limitations: List[str]
async with AsyncMorphik() as db:
response = await db.query(
"Summarize the key research findings from these documents",
filters={"department": "research"},
schema=ResearchFindings
)
# Check if the completion is a dictionary (structured output)
if isinstance(response.completion, dict):
try:
# Parse the dictionary into the Pydantic model
findings = ResearchFindings(**response.completion)
print(f"Main finding: {findings.main_finding}")
print("Supporting evidence:")
for evidence in findings.supporting_evidence:
print(f"- {evidence}")
print("Limitations:")
for limitation in findings.limitations:
print(f"- {limitation}")
except Exception as e:
print(f"Error parsing structured output: {e}")
# Fallback: print the raw dictionary
print(response.completion)
elif isinstance(response.completion, str):
# Fallback to text completion
print(response.completion)