Coverage for pydantic_ai_slim/pydantic_ai/messages.py: 94.27%

1from __future__ import annotations as _annotations (empty)

2 

3import uuid (empty)

4from collections.abc import Sequence (empty)

5from dataclasses import dataclass, field, replace (empty)

6from datetime import datetime (empty)

7from mimetypes import guess_type (empty)

8from typing import Annotated, Any, Literal, Union, cast, overload (empty)

9 

10import pydantic (empty)

11import pydantic_core (empty)

12from opentelemetry._events import Event (empty)

13from typing_extensions import TypeAlias (empty)

14 

15from ._utils import generate_tool_call_id as _generate_tool_call_id, now_utc as _now_utc (empty)

16from .exceptions import UnexpectedModelBehavior (empty)

17 

18 

19@dataclass (empty)

20class SystemPromptPart: (empty)

21 """A system prompt, generally written by the application developer. 

22 

23 This gives the model context and guidance on how to respond. 

24 """ 

25 

26 content: str (empty)

27 """The content of the prompt.""" (empty)

28 

29 timestamp: datetime = field(default_factory=_now_utc) (empty)

30 """The timestamp of the prompt.""" (empty)

31 

32 dynamic_ref: str | None = None (empty)

33 """The ref of the dynamic system prompt function that generated this part. (empty)

34 

35 Only set if system prompt is dynamic, see [`system_prompt`][pydantic_ai.Agent.system_prompt] for more information. 

36 """ 

37 

38 part_kind: Literal['system-prompt'] = 'system-prompt' (empty)

39 """Part type identifier, this is available on all parts as a discriminator.""" (empty)

40 

41 def otel_event(self) -> Event: (empty)

42 return Event('gen_ai.system.message', body={'content': self.content, 'role': 'system'}) (empty)

43 

44 

45@dataclass (empty)

46class AudioUrl: (empty)

47 """A URL to an audio file.""" 

48 

49 url: str (empty)

50 """The URL of the audio file.""" (empty)

51 

52 kind: Literal['audio-url'] = 'audio-url' (empty)

53 """Type identifier, this is available on all parts as a discriminator.""" (empty)

54 

55 @property (empty)

56 def media_type(self) -> AudioMediaType: (empty)

57 """Return the media type of the audio file, based on the url.""" 

58 if self.url.endswith('.mp3'): 

59 return 'audio/mpeg' 

60 elif self.url.endswith('.wav'): 

61 return 'audio/wav' 

62 else: 

63 raise ValueError(f'Unknown audio file extension: {self.url}') 

64 

65 

66@dataclass (empty)

67class ImageUrl: (empty)

68 """A URL to an image.""" 

69 

70 url: str (empty)

71 """The URL of the image.""" (empty)

72 

73 kind: Literal['image-url'] = 'image-url' (empty)

74 """Type identifier, this is available on all parts as a discriminator.""" (empty)

75 

76 @property (empty)

77 def media_type(self) -> ImageMediaType: (empty)

78 """Return the media type of the image, based on the url.""" 

79 if self.url.endswith(('.jpg', '.jpeg')): (empty)

80 return 'image/jpeg' (empty)

81 elif self.url.endswith('.png'): 81 ↛ 82line 81 didn't jump to line 82 because the condition on line 81 was never true(empty)

82 return 'image/png' 

83 elif self.url.endswith('.gif'): 83 ↛ 84line 83 didn't jump to line 84 because the condition on line 83 was never true(empty)

84 return 'image/gif' 

85 elif self.url.endswith('.webp'): 85 ↛ 86line 85 didn't jump to line 86 because the condition on line 85 was never true(empty)

86 return 'image/webp' 

87 else: 

88 raise ValueError(f'Unknown image file extension: {self.url}') (empty)

89 

90 @property (empty)

91 def format(self) -> ImageFormat: (empty)

92 """The file format of the image. 

93 

94 The choice of supported formats were based on the Bedrock Converse API. Other APIs don't require to use a format. 

95 """ 

96 return _image_format(self.media_type) (empty)

97 

98 

99@dataclass (empty)

100class DocumentUrl: (empty)

101 """The URL of the document.""" 

102 

103 url: str (empty)

104 """The URL of the document.""" (empty)

105 

106 kind: Literal['document-url'] = 'document-url' (empty)

107 """Type identifier, this is available on all parts as a discriminator.""" (empty)

108 

109 @property (empty)

110 def media_type(self) -> str: (empty)

111 """Return the media type of the document, based on the url.""" 

112 type_, _ = guess_type(self.url) (empty)

113 if type_ is None: (empty)

114 raise RuntimeError(f'Unknown document file extension: {self.url}') (empty)

115 return type_ (empty)

116 

117 @property (empty)

118 def format(self) -> DocumentFormat: (empty)

119 """The file format of the document. 

120 

121 The choice of supported formats were based on the Bedrock Converse API. Other APIs don't require to use a format. 

122 """ 

123 return _document_format(self.media_type) (empty)

124 

125 

126AudioMediaType: TypeAlias = Literal['audio/wav', 'audio/mpeg'] (empty)

127ImageMediaType: TypeAlias = Literal['image/jpeg', 'image/png', 'image/gif', 'image/webp'] (empty)

128DocumentMediaType: TypeAlias = Literal[ (empty)

129 'application/pdf', 

130 'text/plain', 

131 'text/csv', 

132 'application/vnd.openxmlformats-officedocument.wordprocessingml.document', 

133 'application/vnd.openxmlformats-officedocument.spreadsheetml.sheet', 

134 'text/html', 

135 'text/markdown', 

136 'application/vnd.ms-excel', 

137] 

138AudioFormat: TypeAlias = Literal['wav', 'mp3'] (empty)

139ImageFormat: TypeAlias = Literal['jpeg', 'png', 'gif', 'webp'] (empty)

140DocumentFormat: TypeAlias = Literal['csv', 'doc', 'docx', 'html', 'md', 'pdf', 'txt', 'xls', 'xlsx'] (empty)

141 

142 

143@dataclass (empty)

144class BinaryContent: (empty)

145 """Binary content, e.g. an audio or image file.""" 

146 

147 data: bytes (empty)

148 """The binary data.""" (empty)

149 

150 media_type: AudioMediaType | ImageMediaType | DocumentMediaType | str (empty)

151 """The media type of the binary data.""" (empty)

152 

153 kind: Literal['binary'] = 'binary' (empty)

154 """Type identifier, this is available on all parts as a discriminator.""" (empty)

155 

156 @property (empty)

157 def is_audio(self) -> bool: (empty)

158 """Return `True` if the media type is an audio type.""" 

159 return self.media_type.startswith('audio/') (empty)

160 

161 @property (empty)

162 def is_image(self) -> bool: (empty)

163 """Return `True` if the media type is an image type.""" 

164 return self.media_type.startswith('image/') (empty)

165 

166 @property (empty)

167 def is_document(self) -> bool: (empty)

168 """Return `True` if the media type is a document type.""" 

169 return self.media_type in { (empty)

170 'application/pdf', 

171 'text/plain', 

172 'text/csv', 

173 'application/vnd.openxmlformats-officedocument.wordprocessingml.document', 

174 'application/vnd.openxmlformats-officedocument.spreadsheetml.sheet', 

175 'text/html', 

176 'text/markdown', 

177 'application/vnd.ms-excel', 

178 } 

179 

180 @property (empty)

181 def format(self) -> str: (empty)

182 """The file format of the binary content.""" 

183 if self.is_audio: (empty)

184 if self.media_type == 'audio/mpeg': (empty)

185 return 'mp3' (empty)

186 elif self.media_type == 'audio/wav': 186 ↛ 192line 186 didn't jump to line 192 because the condition on line 186 was always true(empty)

187 return 'wav' (empty)

188 elif self.is_image: (empty)

189 return _image_format(self.media_type) (empty)

190 elif self.is_document: 190 ↛ 192line 190 didn't jump to line 192 because the condition on line 190 was always true(empty)

191 return _document_format(self.media_type) (empty)

192 raise ValueError(f'Unknown media type: {self.media_type}') 

193 

194 

195UserContent: TypeAlias = 'str | ImageUrl | AudioUrl | DocumentUrl | BinaryContent' (empty)

196 

197 

198def _document_format(media_type: str) -> DocumentFormat: (empty)

199 if media_type == 'application/pdf': (empty)

200 return 'pdf' (empty)

201 elif media_type == 'text/plain': (empty)

202 return 'txt' (empty)

203 elif media_type == 'text/csv': (empty)

204 return 'csv' (empty)

205 elif media_type == 'application/vnd.openxmlformats-officedocument.wordprocessingml.document': (empty)

206 return 'docx' (empty)

207 elif media_type == 'application/vnd.openxmlformats-officedocument.spreadsheetml.sheet': (empty)

208 return 'xlsx' (empty)

209 elif media_type == 'text/html': (empty)

210 return 'html' (empty)

211 elif media_type == 'text/markdown': (empty)

212 return 'md' (empty)

213 elif media_type == 'application/vnd.ms-excel': 213 ↛ 216line 213 didn't jump to line 216 because the condition on line 213 was always true(empty)

214 return 'xls' (empty)

215 else: 

216 raise ValueError(f'Unknown document media type: {media_type}') 

217 

218 

219def _image_format(media_type: str) -> ImageFormat: (empty)

220 if media_type == 'image/jpeg': (empty)

221 return 'jpeg' (empty)

222 elif media_type == 'image/png': (empty)

223 return 'png' (empty)

224 elif media_type == 'image/gif': (empty)

225 return 'gif' (empty)

226 elif media_type == 'image/webp': 226 ↛ 229line 226 didn't jump to line 229 because the condition on line 226 was always true(empty)

227 return 'webp' (empty)

228 else: 

229 raise ValueError(f'Unknown image media type: {media_type}') 

230 

231 

232@dataclass (empty)

233class UserPromptPart: (empty)

234 """A user prompt, generally written by the end user. 

235 

236 Content comes from the `user_prompt` parameter of [`Agent.run`][pydantic_ai.Agent.run], 

237 [`Agent.run_sync`][pydantic_ai.Agent.run_sync], and [`Agent.run_stream`][pydantic_ai.Agent.run_stream]. 

238 """ 

239 

240 content: str | Sequence[UserContent] (empty)

241 """The content of the prompt.""" (empty)

242 

243 timestamp: datetime = field(default_factory=_now_utc) (empty)

244 """The timestamp of the prompt.""" (empty)

245 

246 part_kind: Literal['user-prompt'] = 'user-prompt' (empty)

247 """Part type identifier, this is available on all parts as a discriminator.""" (empty)

248 

249 def otel_event(self) -> Event: (empty)

250 if isinstance(self.content, str): (empty)

251 content = self.content (empty)

252 else: 

253 # TODO figure out what to record for images and audio 

254 content = [part if isinstance(part, str) else {'kind': part.kind} for part in self.content] (empty)

255 return Event('gen_ai.user.message', body={'content': content, 'role': 'user'}) (empty)

256 

257 

258tool_return_ta: pydantic.TypeAdapter[Any] = pydantic.TypeAdapter(Any, config=pydantic.ConfigDict(defer_build=True)) (empty)

259 

260 

261@dataclass (empty)

262class ToolReturnPart: (empty)

263 """A tool return message, this encodes the result of running a tool.""" 

264 

265 tool_name: str (empty)

266 """The name of the "tool" was called.""" (empty)

267 

268 content: Any (empty)

269 """The return value.""" (empty)

270 

271 tool_call_id: str (empty)

272 """The tool call identifier, this is used by some models including OpenAI.""" (empty)

273 

274 timestamp: datetime = field(default_factory=_now_utc) (empty)

275 """The timestamp, when the tool returned.""" (empty)

276 

277 part_kind: Literal['tool-return'] = 'tool-return' (empty)

278 """Part type identifier, this is available on all parts as a discriminator.""" (empty)

279 

280 def model_response_str(self) -> str: (empty)

281 """Return a string representation of the content for the model.""" 

282 if isinstance(self.content, str): (empty)

283 return self.content (empty)

284 else: 

285 return tool_return_ta.dump_json(self.content).decode() (empty)

286 

287 def model_response_object(self) -> dict[str, Any]: (empty)

288 """Return a dictionary representation of the content, wrapping non-dict types appropriately.""" 

289 # gemini supports JSON dict return values, but no other JSON types, hence we wrap anything else in a dict 

290 if isinstance(self.content, dict): 290 ↛ 291line 290 didn't jump to line 291 because the condition on line 290 was never true(empty)

291 return tool_return_ta.dump_python(self.content, mode='json') # pyright: ignore[reportUnknownMemberType] 

292 else: 

293 return {'return_value': tool_return_ta.dump_python(self.content, mode='json')} (empty)

294 

295 def otel_event(self) -> Event: (empty)

296 return Event( (empty)

297 'gen_ai.tool.message', 

298 body={'content': self.content, 'role': 'tool', 'id': self.tool_call_id, 'name': self.tool_name}, 

299 ) 

300 

301 

302error_details_ta = pydantic.TypeAdapter(list[pydantic_core.ErrorDetails], config=pydantic.ConfigDict(defer_build=True)) (empty)

303 

304 

305@dataclass (empty)

306class RetryPromptPart: (empty)

307 """A message back to a model asking it to try again. 

308 

309 This can be sent for a number of reasons: 

310 

311 * Pydantic validation of tool arguments failed, here content is derived from a Pydantic 

312 [`ValidationError`][pydantic_core.ValidationError] 

313 * a tool raised a [`ModelRetry`][pydantic_ai.exceptions.ModelRetry] exception 

314 * no tool was found for the tool name 

315 * the model returned plain text when a structured response was expected 

316 * Pydantic validation of a structured response failed, here content is derived from a Pydantic 

317 [`ValidationError`][pydantic_core.ValidationError] 

318 * a result validator raised a [`ModelRetry`][pydantic_ai.exceptions.ModelRetry] exception 

319 """ 

320 

321 content: list[pydantic_core.ErrorDetails] | str (empty)

322 """Details of why and how the model should retry. (empty)

323 

324 If the retry was triggered by a [`ValidationError`][pydantic_core.ValidationError], this will be a list of 

325 error details. 

326 """ 

327 

328 tool_name: str | None = None (empty)

329 """The name of the tool that was called, if any.""" (empty)

330 

331 tool_call_id: str = field(default_factory=_generate_tool_call_id) (empty)

332 """The tool call identifier, this is used by some models including OpenAI. (empty)

333 

334 In case the tool call id is not provided by the model, PydanticAI will generate a random one. 

335 """ 

336 

337 timestamp: datetime = field(default_factory=_now_utc) (empty)

338 """The timestamp, when the retry was triggered.""" (empty)

339 

340 part_kind: Literal['retry-prompt'] = 'retry-prompt' (empty)

341 """Part type identifier, this is available on all parts as a discriminator.""" (empty)

342 

343 def model_response(self) -> str: (empty)

344 """Return a string message describing why the retry is requested.""" 

345 if isinstance(self.content, str): (empty)

346 description = self.content (empty)

347 else: 

348 json_errors = error_details_ta.dump_json(self.content, exclude={'__all__': {'ctx'}}, indent=2) (empty)

349 description = f'{len(self.content)} validation errors: {json_errors.decode()}' (empty)

350 return f'{description}\n\nFix the errors and try again.' (empty)

351 

352 def otel_event(self) -> Event: (empty)

353 if self.tool_name is None: (empty)

354 return Event('gen_ai.user.message', body={'content': self.model_response(), 'role': 'user'}) (empty)

355 else: 

356 return Event( (empty)

357 'gen_ai.tool.message', 

358 body={ 

359 'content': self.model_response(), 

360 'role': 'tool', 

361 'id': self.tool_call_id, 

362 'name': self.tool_name, 

363 }, 

364 ) 

365 

366 

367ModelRequestPart = Annotated[ (empty)

368 Union[SystemPromptPart, UserPromptPart, ToolReturnPart, RetryPromptPart], pydantic.Discriminator('part_kind') 

369] 

370"""A message part sent by PydanticAI to a model.""" (empty)

371 

372 

373@dataclass (empty)

374class ModelRequest: (empty)

375 """A request generated by PydanticAI and sent to a model, e.g. a message from the PydanticAI app to the model.""" 

376 

377 parts: list[ModelRequestPart] (empty)

378 """The parts of the user message.""" (empty)

379 

380 kind: Literal['request'] = 'request' (empty)

381 """Message type identifier, this is available on all parts as a discriminator.""" (empty)

382 

383 

384@dataclass (empty)

385class TextPart: (empty)

386 """A plain text response from a model.""" 

387 

388 content: str (empty)

389 """The text content of the response.""" (empty)

390 

391 part_kind: Literal['text'] = 'text' (empty)

392 """Part type identifier, this is available on all parts as a discriminator.""" (empty)

393 

394 def has_content(self) -> bool: (empty)

395 """Return `True` if the text content is non-empty.""" 

396 return bool(self.content) (empty)

397 

398 

399@dataclass (empty)

400class ToolCallPart: (empty)

401 """A tool call from a model.""" 

402 

403 tool_name: str (empty)

404 """The name of the tool to call.""" (empty)

405 

406 args: str | dict[str, Any] (empty)

407 """The arguments to pass to the tool. (empty)

408 

409 This is stored either as a JSON string or a Python dictionary depending on how data was received. 

410 """ 

411 

412 tool_call_id: str = field(default_factory=_generate_tool_call_id) (empty)

413 """The tool call identifier, this is used by some models including OpenAI. (empty)

414 

415 In case the tool call id is not provided by the model, PydanticAI will generate a random one. 

416 """ 

417 

418 part_kind: Literal['tool-call'] = 'tool-call' (empty)

419 """Part type identifier, this is available on all parts as a discriminator.""" (empty)

420 

421 def args_as_dict(self) -> dict[str, Any]: (empty)

422 """Return the arguments as a Python dictionary. 

423 

424 This is just for convenience with models that require dicts as input. 

425 """ 

426 if isinstance(self.args, dict): (empty)

427 return self.args (empty)

428 args = pydantic_core.from_json(self.args) (empty)

429 assert isinstance(args, dict), 'args should be a dict' (empty)

430 return cast(dict[str, Any], args) (empty)

431 

432 def args_as_json_str(self) -> str: (empty)

433 """Return the arguments as a JSON string. 

434 

435 This is just for convenience with models that require JSON strings as input. 

436 """ 

437 if isinstance(self.args, str): (empty)

438 return self.args (empty)

439 return pydantic_core.to_json(self.args).decode() (empty)

440 

441 def has_content(self) -> bool: (empty)

442 """Return `True` if the arguments contain any data.""" 

443 if isinstance(self.args, dict): (empty)

444 # TODO: This should probably return True if you have the value False, or 0, etc. 

445 # It makes sense to me to ignore empty strings, but not sure about empty lists or dicts 

446 return any(self.args.values()) (empty)

447 else: 

448 return bool(self.args) (empty)

449 

450 

451ModelResponsePart = Annotated[Union[TextPart, ToolCallPart], pydantic.Discriminator('part_kind')] (empty)

452"""A message part returned by a model.""" (empty)

453 

454 

455@dataclass (empty)

456class ModelResponse: (empty)

457 """A response from a model, e.g. a message from the model to the PydanticAI app.""" 

458 

459 parts: list[ModelResponsePart] (empty)

460 """The parts of the model message.""" (empty)

461 

462 model_name: str | None = None (empty)

463 """The name of the model that generated the response.""" (empty)

464 

465 timestamp: datetime = field(default_factory=_now_utc) (empty)

466 """The timestamp of the response. (empty)

467 

468 If the model provides a timestamp in the response (as OpenAI does) that will be used. 

469 """ 

470 

471 kind: Literal['response'] = 'response' (empty)

472 """Message type identifier, this is available on all parts as a discriminator.""" (empty)

473 

474 def otel_events(self) -> list[Event]: (empty)

475 """Return OpenTelemetry events for the response.""" 

476 result: list[Event] = [] (empty)

477 

478 def new_event_body(): (empty)

479 new_body: dict[str, Any] = {'role': 'assistant'} (empty)

480 ev = Event('gen_ai.assistant.message', body=new_body) (empty)

481 result.append(ev) (empty)

482 return new_body (empty)

483 

484 body = new_event_body() (empty)

485 for part in self.parts: (empty)

486 if isinstance(part, ToolCallPart): (empty)

487 body.setdefault('tool_calls', []).append( (empty)

488 { 

489 'id': part.tool_call_id, 

490 'type': 'function', # TODO https://github.com/pydantic/pydantic-ai/issues/888 

491 'function': { 

492 'name': part.tool_name, 

493 'arguments': part.args, 

494 }, 

495 } 

496 ) 

497 elif isinstance(part, TextPart): (empty)

498 if body.get('content'): (empty)

499 body = new_event_body() (empty)

500 body['content'] = part.content (empty)

501 

502 return result (empty)

503 

504 

505ModelMessage = Annotated[Union[ModelRequest, ModelResponse], pydantic.Discriminator('kind')] (empty)

506"""Any message sent to or returned by a model.""" (empty)

507 

508ModelMessagesTypeAdapter = pydantic.TypeAdapter( (empty)

509 list[ModelMessage], config=pydantic.ConfigDict(defer_build=True, ser_json_bytes='base64') 

510) 

511"""Pydantic [`TypeAdapter`][pydantic.type_adapter.TypeAdapter] for (de)serializing messages.""" (empty)

512 

513 

514@dataclass (empty)

515class TextPartDelta: (empty)

516 """A partial update (delta) for a `TextPart` to append new text content.""" 

517 

518 content_delta: str (empty)

519 """The incremental text content to add to the existing `TextPart` content.""" (empty)

520 

521 part_delta_kind: Literal['text'] = 'text' (empty)

522 """Part delta type identifier, used as a discriminator.""" (empty)

523 

524 def apply(self, part: ModelResponsePart) -> TextPart: (empty)

525 """Apply this text delta to an existing `TextPart`. 

526 

527 Args: 

528 part: The existing model response part, which must be a `TextPart`. 

529 

530 Returns: 

531 A new `TextPart` with updated text content. 

532 

533 Raises: 

534 ValueError: If `part` is not a `TextPart`. 

535 """ 

536 if not isinstance(part, TextPart): 536 ↛ 537line 536 didn't jump to line 537 because the condition on line 536 was never true(empty)

537 raise ValueError('Cannot apply TextPartDeltas to non-TextParts') 

538 return replace(part, content=part.content + self.content_delta) (empty)

539 

540 

541@dataclass (empty)

542class ToolCallPartDelta: (empty)

543 """A partial update (delta) for a `ToolCallPart` to modify tool name, arguments, or tool call ID.""" 

544 

545 tool_name_delta: str | None = None (empty)

546 """Incremental text to add to the existing tool name, if any.""" (empty)

547 

548 args_delta: str | dict[str, Any] | None = None (empty)

549 """Incremental data to add to the tool arguments. (empty)

550 

551 If this is a string, it will be appended to existing JSON arguments. 

552 If this is a dict, it will be merged with existing dict arguments. 

553 """ 

554 

555 tool_call_id: str | None = None (empty)

556 """Optional tool call identifier, this is used by some models including OpenAI. (empty)

557 

558 Note this is never treated as a delta — it can replace None, but otherwise if a 

559 non-matching value is provided an error will be raised.""" 

560 

561 part_delta_kind: Literal['tool_call'] = 'tool_call' (empty)

562 """Part delta type identifier, used as a discriminator.""" (empty)

563 

564 def as_part(self) -> ToolCallPart | None: (empty)

565 """Convert this delta to a fully formed `ToolCallPart` if possible, otherwise return `None`. 

566 

567 Returns: 

568 A `ToolCallPart` if both `tool_name_delta` and `args_delta` are set, otherwise `None`. 

569 """ 

570 if self.tool_name_delta is None or self.args_delta is None: (empty)

571 return None (empty)

572 

573 return ToolCallPart(self.tool_name_delta, self.args_delta, self.tool_call_id or _generate_tool_call_id()) (empty)

574 

575 @overload (empty)

576 def apply(self, part: ModelResponsePart) -> ToolCallPart: ... (empty)

577 

578 @overload (empty)

579 def apply(self, part: ModelResponsePart | ToolCallPartDelta) -> ToolCallPart | ToolCallPartDelta: ... (empty)

580 

581 def apply(self, part: ModelResponsePart | ToolCallPartDelta) -> ToolCallPart | ToolCallPartDelta: (empty)

582 """Apply this delta to a part or delta, returning a new part or delta with the changes applied. 

583 

584 Args: 

585 part: The existing model response part or delta to update. 

586 

587 Returns: 

588 Either a new `ToolCallPart` or an updated `ToolCallPartDelta`. 

589 

590 Raises: 

591 ValueError: If `part` is neither a `ToolCallPart` nor a `ToolCallPartDelta`. 

592 UnexpectedModelBehavior: If applying JSON deltas to dict arguments or vice versa. 

593 """ 

594 if isinstance(part, ToolCallPart): (empty)

595 return self._apply_to_part(part) (empty)

596 

597 if isinstance(part, ToolCallPartDelta): 597 ↛ 600line 597 didn't jump to line 600 because the condition on line 597 was always true(empty)

598 return self._apply_to_delta(part) (empty)

599 

600 raise ValueError(f'Can only apply ToolCallPartDeltas to ToolCallParts or ToolCallPartDeltas, not {part}') 

601 

602 def _apply_to_delta(self, delta: ToolCallPartDelta) -> ToolCallPart | ToolCallPartDelta: (empty)

603 """Internal helper to apply this delta to another delta.""" 

604 if self.tool_name_delta: (empty)

605 # Append incremental text to the existing tool_name_delta 

606 updated_tool_name_delta = (delta.tool_name_delta or '') + self.tool_name_delta (empty)

607 delta = replace(delta, tool_name_delta=updated_tool_name_delta) (empty)

608 

609 if isinstance(self.args_delta, str): (empty)

610 if isinstance(delta.args_delta, dict): (empty)

611 raise UnexpectedModelBehavior( (empty)

612 f'Cannot apply JSON deltas to non-JSON tool arguments ({delta=}, {self=})' 

613 ) 

614 updated_args_delta = (delta.args_delta or '') + self.args_delta (empty)

615 delta = replace(delta, args_delta=updated_args_delta) (empty)

616 elif isinstance(self.args_delta, dict): (empty)

617 if isinstance(delta.args_delta, str): (empty)

618 raise UnexpectedModelBehavior( (empty)

619 f'Cannot apply dict deltas to non-dict tool arguments ({delta=}, {self=})' 

620 ) 

621 updated_args_delta = {**(delta.args_delta or {}), **self.args_delta} (empty)

622 delta = replace(delta, args_delta=updated_args_delta) (empty)

623 

624 if self.tool_call_id: (empty)

625 delta = replace(delta, tool_call_id=self.tool_call_id) (empty)

626 

627 # If we now have enough data to create a full ToolCallPart, do so 

628 if delta.tool_name_delta is not None and delta.args_delta is not None: (empty)

629 return ToolCallPart(delta.tool_name_delta, delta.args_delta, delta.tool_call_id or _generate_tool_call_id()) (empty)

630 

631 return delta (empty)

632 

633 def _apply_to_part(self, part: ToolCallPart) -> ToolCallPart: (empty)

634 """Internal helper to apply this delta directly to a `ToolCallPart`.""" 

635 if self.tool_name_delta: (empty)

636 # Append incremental text to the existing tool_name 

637 tool_name = part.tool_name + self.tool_name_delta (empty)

638 part = replace(part, tool_name=tool_name) (empty)

639 

640 if isinstance(self.args_delta, str): (empty)

641 if not isinstance(part.args, str): (empty)

642 raise UnexpectedModelBehavior(f'Cannot apply JSON deltas to non-JSON tool arguments ({part=}, {self=})') (empty)

643 updated_json = part.args + self.args_delta (empty)

644 part = replace(part, args=updated_json) (empty)

645 elif isinstance(self.args_delta, dict): (empty)

646 if not isinstance(part.args, dict): (empty)

647 raise UnexpectedModelBehavior(f'Cannot apply dict deltas to non-dict tool arguments ({part=}, {self=})') (empty)

648 updated_dict = {**(part.args or {}), **self.args_delta} (empty)

649 part = replace(part, args=updated_dict) (empty)

650 

651 if self.tool_call_id: (empty)

652 part = replace(part, tool_call_id=self.tool_call_id) (empty)

653 return part (empty)

654 

655 

656ModelResponsePartDelta = Annotated[Union[TextPartDelta, ToolCallPartDelta], pydantic.Discriminator('part_delta_kind')] (empty)

657"""A partial update (delta) for any model response part.""" (empty)

658 

659 

660@dataclass (empty)

661class PartStartEvent: (empty)

662 """An event indicating that a new part has started. 

663 

664 If multiple `PartStartEvent`s are received with the same index, 

665 the new one should fully replace the old one. 

666 """ 

667 

668 index: int (empty)

669 """The index of the part within the overall response parts list.""" (empty)

670 

671 part: ModelResponsePart (empty)

672 """The newly started `ModelResponsePart`.""" (empty)

673 

674 event_kind: Literal['part_start'] = 'part_start' (empty)

675 """Event type identifier, used as a discriminator.""" (empty)

676 

677 

678@dataclass (empty)

679class PartDeltaEvent: (empty)

680 """An event indicating a delta update for an existing part.""" 

681 

682 index: int (empty)

683 """The index of the part within the overall response parts list.""" (empty)

684 

685 delta: ModelResponsePartDelta (empty)

686 """The delta to apply to the specified part.""" (empty)

687 

688 event_kind: Literal['part_delta'] = 'part_delta' (empty)

689 """Event type identifier, used as a discriminator.""" (empty)

690 

691 

692@dataclass (empty)

693class FinalResultEvent: (empty)

694 """An event indicating the response to the current model request matches the result schema.""" 

695 

696 tool_name: str | None (empty)

697 """The name of the result tool that was called. `None` if the result is from text content and not from a tool.""" (empty)

698 tool_call_id: str | None (empty)

699 """The tool call ID, if any, that this result is associated with.""" (empty)

700 event_kind: Literal['final_result'] = 'final_result' (empty)

701 """Event type identifier, used as a discriminator.""" (empty)

702 

703 

704ModelResponseStreamEvent = Annotated[Union[PartStartEvent, PartDeltaEvent], pydantic.Discriminator('event_kind')] (empty)

705"""An event in the model response stream, either starting a new part or applying a delta to an existing one.""" (empty)

706 

707AgentStreamEvent = Annotated[ (empty)

708 Union[PartStartEvent, PartDeltaEvent, FinalResultEvent], pydantic.Discriminator('event_kind') 

709] 

710"""An event in the agent stream.""" (empty)

711 

712 

713@dataclass (empty)

714class FunctionToolCallEvent: (empty)

715 """An event indicating the start to a call to a function tool.""" 

716 

717 part: ToolCallPart (empty)

718 """The (function) tool call to make.""" (empty)

719 call_id: str = field(init=False) (empty)

720 """An ID used for matching details about the call to its result. If present, defaults to the part's tool_call_id.""" (empty)

721 event_kind: Literal['function_tool_call'] = 'function_tool_call' (empty)

722 """Event type identifier, used as a discriminator.""" (empty)

723 

724 def __post_init__(self): (empty)

725 self.call_id = self.part.tool_call_id or str(uuid.uuid4()) (empty)

726 

727 

728@dataclass (empty)

729class FunctionToolResultEvent: (empty)

730 """An event indicating the result of a function tool call.""" 

731 

732 result: ToolReturnPart | RetryPromptPart (empty)

733 """The result of the call to the function tool.""" (empty)

734 tool_call_id: str (empty)

735 """An ID used to match the result to its original call.""" (empty)

736 event_kind: Literal['function_tool_result'] = 'function_tool_result' (empty)

737 """Event type identifier, used as a discriminator.""" (empty)

738 

739 

740HandleResponseEvent = Annotated[Union[FunctionToolCallEvent, FunctionToolResultEvent], pydantic.Discriminator('kind')] (empty)