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

1from __future__ import annotations as _annotations (empty)

2 

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

4from datetime import datetime (empty)

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

6 

7import pydantic (empty)

8import pydantic_core (empty)

9 

10from ._utils import now_utc as _now_utc (empty)

11from .exceptions import UnexpectedModelBehavior (empty)

12 

13 

14@dataclass (empty)

15class SystemPromptPart: (empty)

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

17 

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

19 """ 

20 

21 content: str (empty)

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

23 

24 dynamic_ref: str | None = None (empty)

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

26 

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

28 """ 

29 

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

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

32 

33 

34@dataclass (empty)

35class UserPromptPart: (empty)

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

37 

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

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

40 """ 

41 

42 content: str (empty)

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

44 

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

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

47 

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

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

50 

51 

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

53 

54 

55@dataclass (empty)

56class ToolReturnPart: (empty)

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

58 

59 tool_name: str (empty)

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

61 

62 content: Any (empty)

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

64 

65 tool_call_id: str | None = None (empty)

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

67 

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

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

70 

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

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

73 

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

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

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

77 return self.content (empty)

78 else: 

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

80 

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

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

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

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

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

86 else: 

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

88 

89 

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

91 

92 

93@dataclass (empty)

94class RetryPromptPart: (empty)

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

96 

97 This can be sent for a number of reasons: 

98 

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

100 [`ValidationError`][pydantic_core.ValidationError] 

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

102 * no tool was found for the tool name 

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

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

105 [`ValidationError`][pydantic_core.ValidationError] 

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

107 """ 

108 

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

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

111 

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

113 error details. 

114 """ 

115 

116 tool_name: str | None = None (empty)

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

118 

119 tool_call_id: str | None = None (empty)

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

121 

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

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

124 

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

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

127 

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

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

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

131 description = self.content (empty)

132 else: 

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

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

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

136 

137 

138ModelRequestPart = Annotated[ (empty)

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

140] 

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

142 

143 

144@dataclass (empty)

145class ModelRequest: (empty)

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

147 

148 parts: list[ModelRequestPart] (empty)

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

150 

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

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

153 

154 

155@dataclass (empty)

156class TextPart: (empty)

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

158 

159 content: str (empty)

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

161 

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

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

164 

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

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

167 return bool(self.content) (empty)

168 

169 

170@dataclass (empty)

171class ToolCallPart: (empty)

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

173 

174 tool_name: str (empty)

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

176 

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

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

179 

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

181 """ 

182 

183 tool_call_id: str | None = None (empty)

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

185 

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

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

188 

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

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

191 

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

193 """ 

194 if isinstance(self.args, dict): 194 ↛ 196line 194 didn't jump to line 196 because the condition on line 194 was always true(empty)

195 return self.args (empty)

196 args = pydantic_core.from_json(self.args) 

197 assert isinstance(args, dict), 'args should be a dict' 

198 return cast(dict[str, Any], args) 

199 

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

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

202 

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

204 """ 

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

206 return self.args (empty)

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

208 

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

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

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

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

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

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

215 else: 

216 return bool(self.args) (empty)

217 

218 

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

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

221 

222 

223@dataclass (empty)

224class ModelResponse: (empty)

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

226 

227 parts: list[ModelResponsePart] (empty)

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

229 

230 model_name: str | None = None (empty)

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

232 

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

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

235 

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

237 """ 

238 

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

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

241 

242 

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

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

245 

246ModelMessagesTypeAdapter = pydantic.TypeAdapter(list[ModelMessage], config=pydantic.ConfigDict(defer_build=True)) (empty)

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

248 

249 

250@dataclass (empty)

251class TextPartDelta: (empty)

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

253 

254 content_delta: str (empty)

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

256 

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

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

259 

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

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

262 

263 Args: 

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

265 

266 Returns: 

267 A new `TextPart` with updated text content. 

268 

269 Raises: 

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

271 """ 

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

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

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

275 

276 

277@dataclass (empty)

278class ToolCallPartDelta: (empty)

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

280 

281 tool_name_delta: str | None = None (empty)

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

283 

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

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

286 

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

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

289 """ 

290 

291 tool_call_id: str | None = None (empty)

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

293 

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

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

296 

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

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

299 

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

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

302 

303 Returns: 

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

305 """ 

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

307 return None (empty)

308 

309 return ToolCallPart( (empty)

310 self.tool_name_delta, 

311 self.args_delta, 

312 self.tool_call_id, 

313 ) 

314 

315 @overload (empty)

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

317 

318 @overload (empty)

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

320 

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

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

323 

324 Args: 

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

326 

327 Returns: 

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

329 

330 Raises: 

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

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

333 """ 

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

335 return self._apply_to_part(part) (empty)

336 

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

338 return self._apply_to_delta(part) (empty)

339 

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

341 

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

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

344 if self.tool_name_delta: (empty)

345 # Append incremental text to the existing tool_name_delta 

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

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

348 

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

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

351 raise UnexpectedModelBehavior( (empty)

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

353 ) 

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

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

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

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

358 raise UnexpectedModelBehavior( (empty)

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

360 ) 

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

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

363 

364 if self.tool_call_id: (empty)

365 # Set the tool_call_id if it wasn't present, otherwise error if it has changed 

366 if delta.tool_call_id is not None and delta.tool_call_id != self.tool_call_id: 366 ↛ 370line 366 didn't jump to line 370 because the condition on line 366 was always true(empty)

367 raise UnexpectedModelBehavior( (empty)

368 f'Cannot apply a new tool_call_id to a ToolCallPartDelta that already has one ({delta=}, {self=})' 

369 ) 

370 delta = replace(delta, tool_call_id=self.tool_call_id) 

371 

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

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

374 return ToolCallPart( (empty)

375 delta.tool_name_delta, 

376 delta.args_delta, 

377 delta.tool_call_id, 

378 ) 

379 

380 return delta (empty)

381 

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

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

384 if self.tool_name_delta: (empty)

385 # Append incremental text to the existing tool_name 

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

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

388 

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

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

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

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

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

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

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

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

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

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

399 

400 if self.tool_call_id: (empty)

401 # Replace the tool_call_id entirely if given 

402 if part.tool_call_id is not None and part.tool_call_id != self.tool_call_id: (empty)

403 raise UnexpectedModelBehavior( (empty)

404 f'Cannot apply a new tool_call_id to a ToolCallPartDelta that already has one ({part=}, {self=})' 

405 ) 

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

407 return part (empty)

408 

409 

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

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

412 

413 

414@dataclass (empty)

415class PartStartEvent: (empty)

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

417 

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

419 the new one should fully replace the old one. 

420 """ 

421 

422 index: int (empty)

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

424 

425 part: ModelResponsePart (empty)

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

427 

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

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

430 

431 

432@dataclass (empty)

433class PartDeltaEvent: (empty)

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

435 

436 index: int (empty)

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

438 

439 delta: ModelResponsePartDelta (empty)

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

441 

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

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

444 

445 

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

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