Implementation:Truera Trulens Dummy App
| Knowledge Sources | |
|---|---|
| Domains | Testing and Instrumentation |
| Last Updated | 2026-02-14 08:00 GMT |
Overview
A dummy/test application that simulates a complete RAG (Retrieval-Augmented Generation) pipeline with retriever, reranker, LLM, template, agent, memory, and tool components, used for testing TruLens instrumentation.
Description
The DummyApp class, defined in examples/dev/dummy_app/app.py, is a test-oriented application that emulates a full RAG pipeline. It inherits from Dummy and composes together several simulated components:
- DummyMemory -- stores queries and answers for conversational context.
- DummyRetriever -- retrieves context chunks given a query.
- DummyReranker -- reranks retrieved chunks by relevance.
- DummyLLM -- simulates language model generation.
- DummyTemplate -- fills prompt templates with question and context.
- DummyTool / DummyStackTool -- simulated tools that process chunks; the first tool records the call stack for debugging.
- DummyAgent -- recursive agents that themselves contain a nested
DummyApp(withnum_agents=0to prevent infinite recursion).
The class provides four main invocation patterns: synchronous (respond_to_query), asynchronous (arespond_to_query), synchronous streaming (stream_respond_to_query), and asynchronous streaming (astream_respond_to_query). Each method runs agents, retrieves and processes context, generates a summary via the LLM, applies a template, and produces a final answer.
All key methods are decorated with @instrument from trulens.apps.app, allowing TruLens to trace and record the execution flow. The app also supports optional parallelism for chunk processing via threads or async tasks.
Usage
This module is used during TruLens development and testing to validate that instrumentation correctly captures calls across a complex, multi-component pipeline. It exercises synchronous, asynchronous, streaming, and parallel execution paths without requiring real LLM or retriever backends.
Code Reference
Source Location
- Repository: Truera_Trulens
- File: examples/dev/dummy_app/app.py
- Lines: 1-391
Signature
class DummyApp(Dummy):
def __init__(
self,
num_agents: int = 2,
num_tools: int = 3,
use_parallel: bool = False,
comp_kwargs: Optional[Dict[Type, Dict[str, Any]]] = None,
**kwargs: Dict[str, Any],
): ...
@instrument
def process_chunk_by_tool(
self, chunk_and_score: Tuple[str, float], tool_num: int = 0
) -> str: ...
@instrument
async def aprocess_chunk_by_tool(
self, chunk_and_score: Tuple[str, float], tool_num: int = 0
) -> str: ...
@instrument
def get_context(self, query: str) -> List[str]: ...
@instrument
async def aget_context(self, query: str) -> List[str]: ...
@instrument
def respond_to_query(self, query: str) -> str: ...
@instrument
async def arespond_to_query(self, query: str) -> str: ...
@instrument
def stream_respond_to_query(self, query: str) -> Iterable[str]: ...
@instrument
async def astream_respond_to_query(self, query: str) -> AsyncIterable[str]: ...
Import
from examples.dev.dummy_app.app import DummyApp
I/O Contract
Constructor Parameters
| Parameter | Type | Default | Description |
|---|---|---|---|
num_agents |
int |
2 |
Number of recursive DummyAgent instances to create. |
num_tools |
int |
3 |
Number of tools to create (minimum 1). The first is always a DummyStackTool. |
use_parallel |
bool |
False |
Whether to use parallelism (threads or async) for processing retrieval chunks. |
comp_kwargs |
Optional[Dict[Type, Dict[str, Any]]] |
None |
Dictionary mapping component classes to keyword arguments for their constructors. |
**kwargs |
Dict[str, Any] |
-- | Additional arguments passed to all component constructors via the Dummy base class. |
Key Method: respond_to_query
| Direction | Name | Type | Description |
|---|---|---|---|
| Input | query |
str |
The user query to respond to. |
| Output | return | str |
The final generated answer string. |
Key Method: get_context
| Direction | Name | Type | Description |
|---|---|---|---|
| Input | query |
str |
The query to retrieve and process context for. |
| Output | return | List[str] |
List of processed context chunks after retrieval, reranking, and tool processing. |
Internal Components
| Component | Type | Description |
|---|---|---|
self.memory |
DummyMemory |
Stores queries and answers for memory recall. |
self.retriever |
DummyRetriever |
Retrieves context chunks for a given query. |
self.llm |
DummyLLM |
Generates text from input strings. |
self.template |
DummyTemplate |
Fills prompt templates with question and context. |
self.tools |
List[DummyTool] |
List of tools for chunk processing (first is DummyStackTool). |
self.agents |
List[DummyAgent] |
Recursive agents that each contain a nested DummyApp. |
self.reranker |
DummyReranker |
Reranks retrieved chunks by relevance. |
Usage Examples
from examples.dev.dummy_app.app import DummyApp
# Create a basic DummyApp with default settings
app = DummyApp()
# Synchronous query
answer = app.respond_to_query("What is the capital of France?")
# Asynchronous query
import asyncio
answer = asyncio.run(app.arespond_to_query("What is the capital of France?"))
# Streaming response
for chunk in app.stream_respond_to_query("Explain gravity."):
print(chunk, end="")
# Create with parallelism and custom tool count
parallel_app = DummyApp(num_tools=5, num_agents=1, use_parallel=True)
answer = parallel_app.respond_to_query("Summarize the document.")
# Retrieve context only
context_chunks = app.get_context("What is machine learning?")