query

Generate completion using relevant chunks as context

query

Generate completion using relevant chunks as context.

• Sync
• 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

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

• Sync
• Async

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

• Sync
• Async

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

• Sync
• Async

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.

• Sync
• Async

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)