InMemoryRunner that is tailored for tests.
| 243 | |
| 244 | |
| 245 | class InMemoryRunner: |
| 246 | """InMemoryRunner that is tailored for tests.""" |
| 247 | |
| 248 | def __init__( |
| 249 | self, |
| 250 | root_agent: Optional[Agent | LlmAgent] = None, |
| 251 | response_modalities: list[str] = None, |
| 252 | plugins: list[BasePlugin] = [], |
| 253 | app: Optional[App] = None, |
| 254 | node: Any = None, |
| 255 | ): |
| 256 | """Initializes the InMemoryRunner. |
| 257 | |
| 258 | Args: |
| 259 | root_agent: The root agent to run, won't be used if app is provided. |
| 260 | response_modalities: The response modalities of the runner. |
| 261 | plugins: The plugins to use in the runner, won't be used if app is |
| 262 | provided. |
| 263 | app: The app to use in the runner. |
| 264 | node: The root node to run. |
| 265 | """ |
| 266 | self._app = app |
| 267 | if node: |
| 268 | self.app_name = node.name |
| 269 | self.root_agent = None |
| 270 | self.runner = Runner( |
| 271 | node=node, |
| 272 | artifact_service=InMemoryArtifactService(), |
| 273 | session_service=InMemorySessionService(), |
| 274 | memory_service=InMemoryMemoryService(), |
| 275 | plugins=plugins, |
| 276 | ) |
| 277 | elif not app: |
| 278 | self.app_name = 'test_app' |
| 279 | self.root_agent = root_agent |
| 280 | self.runner = Runner( |
| 281 | app_name='test_app', |
| 282 | agent=root_agent, |
| 283 | artifact_service=InMemoryArtifactService(), |
| 284 | session_service=InMemorySessionService(), |
| 285 | memory_service=InMemoryMemoryService(), |
| 286 | plugins=plugins, |
| 287 | ) |
| 288 | else: |
| 289 | self.app_name = app.name |
| 290 | self.root_agent = app.root_agent |
| 291 | self.runner = Runner( |
| 292 | app=app, |
| 293 | artifact_service=InMemoryArtifactService(), |
| 294 | session_service=InMemorySessionService(), |
| 295 | memory_service=InMemoryMemoryService(), |
| 296 | ) |
| 297 | self.session_id = None |
| 298 | |
| 299 | @property |
| 300 | def session(self) -> Session: |
| 301 | if not self.session_id: |
| 302 | session = self.runner.session_service.create_session_sync( |
no outgoing calls