rag

• __all__ = ['RagContext', 'RagEngine'] module-attribute

Bases: SerializableMixin “)

Attributes

Name	Type	Description
<a href="./#griptape.engines.rag.RagContext.query" title="query = field(metadata={'serializable': True}) class-attribute instance-attribute (griptape.engines.rag.RagContext.query)"><code class="language-python">query</code></a>	str	Query provided by the user.<br/>
<a href="./#griptape.engines.rag.RagContext.module_configs" title="module_configs = field(factory=dict, metadata={'serializable': True}) class-attribute instance-attribute (griptape.engines.rag.RagContext.module_configs)"><code class="language-python">module_configs</code></a>	dict[str, dict]	Dictionary of module configs. First key should be a module name and the second a dictionary of configs parameters.<br/>
<a href="./#griptape.engines.rag.RagContext.before_query" title="before_query = field(factory=list, metadata={'serializable': True}) class-attribute instance-attribute (griptape.engines.rag.RagContext.before_query)"><code class="language-python">before_query</code></a>	<a href="../../../data/artifacts.mdx#list" title="List (list)"><code class="language-python">list[str]</code></a>	An optional list of strings to add before the query in response modules.<br/>
<a href="./#griptape.engines.rag.RagContext.after_query" title="after_query = field(factory=list, metadata={'serializable': True}) class-attribute instance-attribute (griptape.engines.rag.RagContext.after_query)"><code class="language-python">after_query</code></a>	<a href="../../../data/artifacts.mdx#list" title="List (list)"><code class="language-python">list[str]</code></a>	An optional list of strings to add after the query in response modules.<br/>
<a href="./#griptape.engines.rag.RagContext.text_chunks" title="text_chunks = field(factory=list, metadata={'serializable': True}) class-attribute instance-attribute (griptape.engines.rag.RagContext.text_chunks)"><code class="language-python">text_chunks</code></a>	<a href="../../../data/artifacts.mdx#list" title="List (list)"><code class="language-python">list[TextArtifact]</code></a>	A list of text chunks to pass around from the retrieval stage to the response stage.<br/>
<a href="./#griptape.engines.rag.RagContext.outputs" title="outputs = field(factory=list, metadata={'serializable': True}) class-attribute instance-attribute (griptape.engines.rag.RagContext.outputs)"><code class="language-python">outputs</code></a>	<a href="../../../data/artifacts.mdx#list" title="List (list)"><code class="language-python">list[BaseArtifact]</code></a>	List of outputs from the response stage.<br/>

Source Code in griptape/engines/rag/rag_context.py

@define(kw_only=True)
class RagContext(SerializableMixin):
    """Used by RagEngine stages and module to pass context that individual modules are expected to update in the `run` method.

    Attributes:
        query: Query provided by the user.
        module_configs: Dictionary of module configs. First key should be a module name and the second a dictionary of configs parameters.
        before_query: An optional list of strings to add before the query in response modules.
        after_query: An optional list of strings to add after the query in response modules.
        text_chunks: A list of text chunks to pass around from the retrieval stage to the response stage.
        outputs: List of outputs from the response stage.
    """

    query: str = field(metadata={"serializable": True})
    module_configs: dict[str, dict] = field(factory=dict, metadata={"serializable": True})
    before_query: list[str] = field(factory=list, metadata={"serializable": True})
    after_query: list[str] = field(factory=list, metadata={"serializable": True})
    text_chunks: list[TextArtifact] = field(factory=list, metadata={"serializable": True})
    outputs: list[BaseArtifact] = field(factory=list, metadata={"serializable": True})

    def get_references(self) -> list[Reference]:
        return utils.references_from_artifacts(self.text_chunks)

• after_query = field(factory=list, metadata={'serializable': True}) class-attribute instance-attribute

• before_query = field(factory=list, metadata={'serializable': True}) class-attribute instance-attribute

• module_configs = field(factory=dict, metadata={'serializable': True}) class-attribute instance-attribute

• outputs = field(factory=list, metadata={'serializable': True}) class-attribute instance-attribute

• query = field(metadata={'serializable': True}) class-attribute instance-attribute

• text_chunks = field(factory=list, metadata={'serializable': True}) class-attribute instance-attribute

get_references()

Source Code in griptape/engines/rag/rag_context.py

def get_references(self) -> list[Reference]:
    return utils.references_from_artifacts(self.text_chunks)

RagEngine

Source Code in griptape/engines/rag/rag_engine.py

@define(kw_only=True)
class RagEngine:
    query_stage: Optional[QueryRagStage] = field(default=None)
    retrieval_stage: Optional[RetrievalRagStage] = field(default=None)
    response_stage: Optional[ResponseRagStage] = field(default=None)

    def __attrs_post_init__(self) -> None:
        modules = []

        if self.query_stage is not None:
            modules.extend(self.query_stage.modules)

        if self.retrieval_stage is not None:
            modules.extend(self.retrieval_stage.modules)

        if self.response_stage is not None:
            modules.extend(self.response_stage.modules)

        module_names = [m.name for m in modules]

        if len(module_names) > len(set(module_names)):
            raise ValueError("module names have to be unique")

    def process_query(self, query: str) -> RagContext:
        return self.process(RagContext(query=query))

    def process(self, context: RagContext) -> RagContext:
        if self.query_stage:
            context = self.query_stage.run(context)

        if self.retrieval_stage:
            context = self.retrieval_stage.run(context)

        if self.response_stage:
            context = self.response_stage.run(context)

        return context

• query_stage = field(default=None) class-attribute instance-attribute

• response_stage = field(default=None) class-attribute instance-attribute

• retrieval_stage = field(default=None) class-attribute instance-attribute

attrs_post_init()

Source Code in griptape/engines/rag/rag_engine.py

def __attrs_post_init__(self) -> None:
    modules = []

    if self.query_stage is not None:
        modules.extend(self.query_stage.modules)

    if self.retrieval_stage is not None:
        modules.extend(self.retrieval_stage.modules)

    if self.response_stage is not None:
        modules.extend(self.response_stage.modules)

    module_names = [m.name for m in modules]

    if len(module_names) > len(set(module_names)):
        raise ValueError("module names have to be unique")

process(context)

Source Code in griptape/engines/rag/rag_engine.py

def process(self, context: RagContext) -> RagContext:
    if self.query_stage:
        context = self.query_stage.run(context)

    if self.retrieval_stage:
        context = self.retrieval_stage.run(context)

    if self.response_stage:
        context = self.response_stage.run(context)

    return context

process_query(query)

Source Code in griptape/engines/rag/rag_engine.py

def process_query(self, query: str) -> RagContext:
    return self.process(RagContext(query=query))

• modules
  • run(context)abstractmethod
    • BaseBeforeResponseRagModule
  • run(context)abstractmethod
    • BaseQueryRagModule
  • run(context)abstractmethod
    • BaseRagModule
  • generate_prompt_stack(system_prompt, query)
  • get_context_param(context, key)
  • set_context_param(context, key, value)
    • BaseRerankRagModule
  • run(context)abstractmethod
    • BaseResponseRagModule
  • run(context)abstractmethod
    • BaseRetrievalRagModule
  • run(context)abstractmethod
    • FootnotePromptResponseRagModule
  • default_generate_system_template(context, artifacts)
    • PromptResponseRagModule
  • default_generate_system_template(context, artifacts)
  • run(context)
    • TextChunksRerankRagModule
  • run(context)
    • TextChunksResponseRagModule
  • run(context)
    • TextLoaderRetrievalRagModule
  • run(context)
    • TranslateQueryRagModule
  • default_generate_user_template(query, language)
  • run(context)
    • VectorStoreRetrievalRagModule
  • run(context)
    • base_rag_module
    • query
    • response
    • retrieval
• rag_context
  • Attributes
  • get_references()
• rag_engine
  • attrs_post_init()
  • process(context)
  • process_query(query)
• stages
  • run(context)abstractmethod
    • QueryRagStage
  • run(context)
    • ResponseRagStage
  • run(context)
    • RetrievalRagStage
  • run(context)
    • base_rag_stage
    • query_rag_stage
    • response_rag_stage
    • retrieval_rag_stage