Skip to content

Framework Integrations

Only official framework adapters are accepted by Optimizer. Install the matching extra: langchain, langgraph, or google-adk (Python 3.10+). Factories must return real framework objects. Mapping functions configure framework inputs, configuration, session state, and outputs.

tunarag.integrations.google_adk.GoogleADKAdapter 

GoogleADKAdapter(runner: Any = None, *, factory: Callable[[Candidate], Any] | None = None, state_mapper: Callable[[Candidate], Mapping[str, Any]] | None = None, output_mapper: Callable[[Any], Any] | None = None, identity: Any = None)

Evaluate a Google ADK runner in a fresh session for every sample and retry.

Candidate parameters become initial session state. Use a factory when changing an agent's model, tools, or graph requires constructing a new framework runner.

Source code in src/tunarag/integrations/google_adk.py 
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
def __init__(
    self,
    runner: Any = None,
    *,
    factory: Callable[[Candidate], Any] | None = None,
    state_mapper: Callable[[Candidate], Mapping[str, Any]] | None = None,
    output_mapper: Callable[[Any], Any] | None = None,
    identity: Any = None,
) -> None:
    if (runner is None) == (factory is None):
        raise ValueError("provide exactly one of runner or factory")
    for name, callback in (
        ("factory", factory),
        ("state_mapper", state_mapper),
        ("output_mapper", output_mapper),
    ):
        if callback is not None and not callable(callback):
            raise TypeError(f"{name} must be callable")
    if runner is not None:
        _require_runner(runner)
    self._runner = runner
    self._factory = factory
    self._state_mapper = state_mapper
    self._output_mapper = output_mapper
    self._identity = identity

run async 

run(candidate: Candidate, example: Any) -> tuple[Any, tuple[UsageRecord, ...]]

Consume runner events, capture usage, and remove the temporary session.

Source code in src/tunarag/integrations/google_adk.py 
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
async def run(self, candidate: Candidate, example: Any) -> tuple[Any, tuple[UsageRecord, ...]]:
    """Consume runner events, capture usage, and remove the temporary session."""
    runner = self._runner if self._factory is None else self._factory(candidate)
    _require_runner(runner)
    types = import_module("google.genai.types")
    query = getattr(example, "query", example)
    if not isinstance(query, str) or not query.strip():
        raise TypeError("Google ADK evaluation input must contain a nonempty query")
    state = (
        dict(candidate.parameters)
        if self._state_mapper is None
        else self._state_mapper(candidate)
    )
    if not isinstance(state, Mapping):
        raise TypeError("state_mapper must return a mapping")
    session_id = uuid4().hex
    user_id = "tunarag-evaluation"
    session = await runner.session_service.create_session(
        app_name=runner.app_name,
        user_id=user_id,
        session_id=session_id,
        state=dict(state),
    )
    usage: list[UsageRecord] = []
    final_event = None
    started_at = time.perf_counter()
    events = runner.run_async(
        user_id=user_id,
        session_id=session.id,
        new_message=types.Content(role="user", parts=[types.Part.from_text(text=query)]),
    )
    try:
        async for event in events:
            if event.error_code:
                raise AdapterError("Google ADK runner returned an error event", stage="adapter")
            metadata = event.usage_metadata
            if metadata is not None:
                usage.append(
                    UsageRecord(
                        "google_adk.model",
                        input_tokens=_token_count(metadata.prompt_token_count),
                        output_tokens=_token_count(metadata.candidates_token_count),
                    )
                )
            if event.is_final_response():
                final_event = event
        if final_event is None:
            raise AdapterError("Google ADK runner produced no final response", stage="adapter")
        if self._output_mapper is None and final_event.content is None:
            raise AdapterError(
                "ADK final event has no text; provide output_mapper for structured output",
                stage="adapter",
            )
        output = (
            self._output_mapper(final_event)
            if self._output_mapper is not None
            else {
                "answer": "".join(
                    part.text or ""
                    for part in final_event.content.parts or ()
                    if not part.thought
                )
            }
        )
        usage.append(
            UsageRecord("google_adk.runner", latency_seconds=time.perf_counter() - started_at)
        )
        return output, tuple(usage)
    finally:
        try:
            await events.aclose()
        finally:
            await runner.session_service.delete_session(
                app_name=runner.app_name, user_id=user_id, session_id=session.id
            )

fingerprint 

fingerprint() -> dict[str, Any]

Describe integration semantics without sessions or provider credentials.

Source code in src/tunarag/integrations/google_adk.py 
125
126
127
128
129
130
131
132
133
134
135
def fingerprint(self) -> dict[str, Any]:
    """Describe integration semantics without sessions or provider credentials."""
    return {
        "integration": "google_adk",
        "schema_version": 1,
        "runner": _type_name(self._runner) if self._runner is not None else None,
        "factory": _callable_identity(self._factory),
        "state_mapper": _callable_identity(self._state_mapper),
        "output_mapper": _callable_identity(self._output_mapper),
        "identity": self._identity,
    }

Google ADK creates and deletes a fresh session for each sample, including retries. Candidate parameters become session state; agent construction settings require a runner factory. The application owns and closes runners. For LangGraph checkpointing, map each evaluation to its own thread instead of reusing a production conversation. Include model, corpus, prompt, and mapper revisions in adapter identity so cache and resume do not mix pipeline versions.

LangChain and LangGraph

runnables

Adapters for LangChain runnables and compiled LangGraph graphs.

RunnableFactory module-attribute 

RunnableFactory = Callable[[Candidate], Any]

RunnableInputMapper module-attribute 

RunnableInputMapper = Callable[[Any, Candidate], Any]

RunnableConfigMapper module-attribute 

RunnableConfigMapper = Callable[[Candidate], Mapping[str, Any] | None]

RunnableOutputMapper module-attribute 

RunnableOutputMapper = Callable[[Any], Any]

RunnableUsageMapper module-attribute 

RunnableUsageMapper = Callable[[Any], Sequence[UsageRecord]]

LangChainAdapter 

LangChainAdapter(runnable: Any = None, *, factory: RunnableFactory | None = None, input_key: str = 'question', input_mapper: RunnableInputMapper | None = None, config_mapper: RunnableConfigMapper | None = None, output_mapper: RunnableOutputMapper | None = None, usage_mapper: RunnableUsageMapper | None = None, capture_message_usage: bool = True, identity: Any = None)

Bases: _RunnableAdapter

Expose a LangChain runnable through TunaRAG's standard adapter contract.

Source code in src/tunarag/integrations/runnables.py 
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
def __init__(
    self,
    runnable: Any = None,
    *,
    factory: RunnableFactory | None = None,
    input_key: str = "question",
    input_mapper: RunnableInputMapper | None = None,
    config_mapper: RunnableConfigMapper | None = None,
    output_mapper: RunnableOutputMapper | None = None,
    usage_mapper: RunnableUsageMapper | None = None,
    capture_message_usage: bool = True,
    identity: Any = None,
) -> None:
    super().__init__(
        runnable,
        factory=factory,
        input_key=input_key,
        input_mapper=input_mapper,
        config_mapper=config_mapper,
        output_mapper=output_mapper,
        usage_mapper=usage_mapper,
        capture_message_usage=capture_message_usage,
        identity=identity,
        framework="langchain",
    )

LangGraphAdapter 

LangGraphAdapter(graph: Any = None, *, factory: RunnableFactory | None = None, input_key: str = 'question', input_mapper: RunnableInputMapper | None = None, config_mapper: RunnableConfigMapper | None = None, output_mapper: RunnableOutputMapper | None = None, usage_mapper: RunnableUsageMapper | None = None, capture_message_usage: bool = True, identity: Any = None)

Bases: _RunnableAdapter

Expose a compiled LangGraph graph through TunaRAG's adapter contract.

Source code in src/tunarag/integrations/runnables.py 
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
def __init__(
    self,
    graph: Any = None,
    *,
    factory: RunnableFactory | None = None,
    input_key: str = "question",
    input_mapper: RunnableInputMapper | None = None,
    config_mapper: RunnableConfigMapper | None = None,
    output_mapper: RunnableOutputMapper | None = None,
    usage_mapper: RunnableUsageMapper | None = None,
    capture_message_usage: bool = True,
    identity: Any = None,
) -> None:
    super().__init__(
        graph,
        factory=factory,
        input_key=input_key,
        input_mapper=input_mapper,
        config_mapper=config_mapper,
        output_mapper=output_mapper,
        usage_mapper=usage_mapper,
        capture_message_usage=capture_message_usage,
        identity=identity,
        framework="langgraph",
    )

MLflow

mlflow

Optional post-commit MLflow tracking mirror.

MLflowClient

Bases: Protocol

Subset of the low-level MLflow client used by the mirror.

create_run 

create_run(experiment_id: str, *, tags: Mapping[str, str]) -> Any
Source code in src/tunarag/integrations/mlflow.py 
21
def create_run(self, experiment_id: str, *, tags: Mapping[str, str]) -> Any: ...

log_metric 

log_metric(run_id: str, key: str, value: float, *, step: int | None = None) -> Any
Source code in src/tunarag/integrations/mlflow.py 
23
24
25
def log_metric(
    self, run_id: str, key: str, value: float, *, step: int | None = None
) -> Any: ...

log_param 

log_param(run_id: str, key: str, value: Any) -> Any
Source code in src/tunarag/integrations/mlflow.py 
27
def log_param(self, run_id: str, key: str, value: Any) -> Any: ...

set_tag 

set_tag(run_id: str, key: str, value: Any) -> None
Source code in src/tunarag/integrations/mlflow.py 
29
def set_tag(self, run_id: str, key: str, value: Any) -> None: ...

set_terminated 

set_terminated(run_id: str, *, status: str | None = None) -> None
Source code in src/tunarag/integrations/mlflow.py 
31
def set_terminated(self, run_id: str, *, status: str | None = None) -> None: ...

MLflowCallback 

MLflowCallback(store: SQLiteStore, *, experiment_id: str = '0', tracking_uri: str | None = None, tags: Mapping[str, str] | None = None, client: MLflowClient | None = None)

Mirror committed study events into one low-level MLflow run.

Source code in src/tunarag/integrations/mlflow.py 
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
def __init__(
    self,
    store: SQLiteStore,
    *,
    experiment_id: str = "0",
    tracking_uri: str | None = None,
    tags: Mapping[str, str] | None = None,
    client: MLflowClient | None = None,
) -> None:
    if not experiment_id.strip():
        raise ValueError("MLflow experiment id must not be empty")
    self._store = store
    self._experiment_id = experiment_id
    self._tracking_uri = tracking_uri
    safe_tags = redact(tags or {})
    if not isinstance(safe_tags, Mapping):
        raise TypeError("MLflow tags must be a mapping")
    self._tags = {str(key): str(value) for key, value in safe_tags.items()}
    self._client = client
    self._run_ids: dict[str, str] = {}

on_event async 

on_event(event: StudyEvent) -> None

Mirror one durable event after gathering committed local state.

Source code in src/tunarag/integrations/mlflow.py 
66
67
68
69
70
async def on_event(self, event: StudyEvent) -> None:
    """Mirror one durable event after gathering committed local state."""

    trial_data = self._trial_data(event)
    await asyncio.to_thread(self._mirror, event, trial_data)