diff --git a/dspy/__init__.py b/dspy/__init__.py index 76afdb805d..264f6af39d 100644 --- a/dspy/__init__.py +++ b/dspy/__init__.py @@ -6,7 +6,7 @@ from dspy.evaluate import Evaluate # isort: skip from dspy.clients import * # isort: skip -from dspy.adapters import Adapter, ChatAdapter, JSONAdapter, XMLAdapter, TwoStepAdapter, Image, Audio, File, History, Type, Tool, ToolCalls, Code, Reasoning # isort: skip +from dspy.adapters import Adapter, ChatAdapter, JSONAdapter, XMLAdapter, TwoStepAdapter, Image, Audio, File, History, HistoryFrame, Observation, Type, Tool, ToolCalls, Code, Reasoning # isort: skip from dspy.primitives.sandbox_serializable import SandboxSerializable # isort: skip from dspy.utils.exceptions import ContextWindowExceededError from dspy.utils.logging_utils import configure_dspy_loggers, disable_logging, enable_logging diff --git a/dspy/adapters/__init__.py b/dspy/adapters/__init__.py index c217d7260e..7b0270f614 100644 --- a/dspy/adapters/__init__.py +++ b/dspy/adapters/__init__.py @@ -2,7 +2,19 @@ from dspy.adapters.chat_adapter import ChatAdapter from dspy.adapters.json_adapter import JSONAdapter from dspy.adapters.two_step_adapter import TwoStepAdapter -from dspy.adapters.types import Audio, Code, File, History, Image, Reasoning, Tool, ToolCalls, Type +from dspy.adapters.types import ( + Audio, + Code, + File, + History, + HistoryFrame, + Image, + Observation, + Reasoning, + Tool, + ToolCalls, + Type, +) from dspy.adapters.xml_adapter import XMLAdapter __all__ = [ @@ -10,6 +22,8 @@ "ChatAdapter", "Type", "History", + "HistoryFrame", + "Observation", "Image", "Audio", "File", diff --git a/dspy/adapters/types/__init__.py b/dspy/adapters/types/__init__.py index 5ec8043021..aaa0e8b86d 100644 --- a/dspy/adapters/types/__init__.py +++ b/dspy/adapters/types/__init__.py @@ -2,9 +2,21 @@ from dspy.adapters.types.base_type import Type from dspy.adapters.types.code import Code from dspy.adapters.types.file import File -from dspy.adapters.types.history import History +from dspy.adapters.types.history import History, HistoryFrame, Observation from dspy.adapters.types.image import Image from dspy.adapters.types.reasoning import Reasoning from dspy.adapters.types.tool import Tool, ToolCalls -__all__ = ["History", "Image", "Audio", "File", "Type", "Tool", "ToolCalls", "Code", "Reasoning"] +__all__ = [ + "History", + "HistoryFrame", + "Observation", + "Image", + "Audio", + "File", + "Type", + "Tool", + "ToolCalls", + "Code", + "Reasoning", +] diff --git a/dspy/adapters/types/history.py b/dspy/adapters/types/history.py index 6dda4f9b7c..5d70d3894d 100644 --- a/dspy/adapters/types/history.py +++ b/dspy/adapters/types/history.py @@ -1,68 +1,140 @@ -from typing import Any +from typing import Any, Callable import pydantic +from pydantic import Field, model_validator -class History(pydantic.BaseModel): - """Class representing the conversation history. - - The conversation history is a list of messages, each message entity should have keys from the associated signature. - For example, if you have the following signature: - - ``` - class MySignature(dspy.Signature): - question: str = dspy.InputField() - history: dspy.History = dspy.InputField() - answer: str = dspy.OutputField() - ``` - - Then the history should be a list of dictionaries with keys "question" and "answer". +class Observation(pydantic.BaseModel): + """External result produced by executing or evaluating a history frame.""" - Examples: - ``` - import dspy + value: Any + source: str | None = None + call_id: str | None = None + name: str | None = None + is_error: bool = False - dspy.configure(lm=dspy.LM("openai/gpt-4o-mini")) - class MySignature(dspy.Signature): - question: str = dspy.InputField() - history: dspy.History = dspy.InputField() - answer: str = dspy.OutputField() +class HistoryFrame(pydantic.BaseModel): + """A partial DSPy example/prediction frame with optional observations.""" - history = dspy.History( - messages=[ - {"question": "What is the capital of France?", "answer": "Paris"}, - {"question": "What is the capital of Germany?", "answer": "Berlin"}, - ] - ) + inputs: dict[str, Any] = Field(default_factory=dict) + outputs: dict[str, Any] = Field(default_factory=dict) + observations: list[Observation] = Field(default_factory=list) + complete: bool = False + source: str | None = None - predict = dspy.Predict(MySignature) - outputs = predict(question="What is the capital of France?", history=history) - ``` + model_config = pydantic.ConfigDict(extra="forbid") - Example of capturing the conversation history: - ``` - import dspy - dspy.configure(lm=dspy.LM("openai/gpt-4o-mini")) +HistoryEntry = HistoryFrame | dict[str, Any] - class MySignature(dspy.Signature): - question: str = dspy.InputField() - history: dspy.History = dspy.InputField() - answer: str = dspy.OutputField() - predict = dspy.Predict(MySignature) - outputs = predict(question="What is the capital of France?") - history = dspy.History(messages=[{"question": "What is the capital of France?", **outputs}]) - outputs_with_history = predict(question="Are you sure?", history=history) - ``` - """ +class History(pydantic.BaseModel): + """Reusable DSPy field-frame history.""" - messages: list[dict[str, Any]] + frames: list[HistoryEntry] = Field(default_factory=list) model_config = pydantic.ConfigDict( - frozen=True, str_strip_whitespace=True, - validate_assignment=True, extra="forbid", ) + + @model_validator(mode="before") + @classmethod + def _accept_legacy_messages_key(cls, data: Any) -> Any: + if isinstance(data, dict) and "messages" in data and "frames" not in data: + data = dict(data) + data["frames"] = data.pop("messages") + return data + + def __init__(self, *args: Any, compact_fn: Callable[["History"], None] | None = None, **kwargs: Any): + super().__init__(*args, **kwargs) + object.__setattr__(self, "_compact_fn", compact_fn) + + @property + def messages(self) -> list[HistoryEntry]: + return self.frames + + def compact_if_needed(self) -> None: + fn = getattr(self, "_compact_fn", None) + if fn is not None: + fn(self) + + def append_inputs(self, inputs: dict[str, Any], *, source: str | None = None) -> HistoryFrame: + frame = HistoryFrame(inputs=dict(inputs), source=source) + self.frames.append(frame) + return frame + + def append_outputs( + self, + outputs: dict[str, Any], + *, + observations: list[Observation] | None = None, + complete: bool = False, + source: str | None = None, + ) -> HistoryFrame: + frame = HistoryFrame( + outputs=dict(outputs), + observations=list(observations or []), + complete=complete, + source=source, + ) + self.frames.append(frame) + return frame + + def append_observation( + self, + value: Any, + *, + source: str | None = None, + call_id: str | None = None, + name: str | None = None, + is_error: bool = False, + ) -> Observation: + observation = Observation(value=value, source=source, call_id=call_id, name=name, is_error=is_error) + if self.frames and isinstance(self.frames[-1], HistoryFrame): + self.frames[-1].observations.append(observation) + else: + self.frames.append(HistoryFrame(observations=[observation], source=source)) + return observation + + def append_input(self, inputs: dict[str, Any]) -> HistoryFrame: + return self.append_inputs(inputs) + + def append_output(self, outputs: dict[str, Any]) -> HistoryFrame: + return self.append_outputs(outputs, complete=True) + + def has_open_episode(self) -> bool: + last_boundary = None + for frame in self.frames: + if isinstance(frame, dict): + continue + if frame.inputs: + last_boundary = "input" + if frame.complete: + last_boundary = "output" + return last_boundary == "input" + + +def truncate_oldest_actions(history: History, *, max_tokens: int = 200_000, keep_n: int = 3) -> None: + if len(str(history.frames)) // 4 <= max_tokens: + return + + action_starts = [ + idx + for idx, frame in enumerate(history.frames) + if isinstance(frame, HistoryFrame) and frame.observations + ] + drop_count = len(action_starts) - keep_n + if drop_count <= 0: + return + + drop_indices = set(action_starts[:drop_count]) + history.frames[:] = [frame for idx, frame in enumerate(history.frames) if idx not in drop_indices] + + +def make_truncate_oldest_actions(max_tokens: int = 200_000, keep_n: int = 3) -> Callable[[History], None]: + def _compact(history: History) -> None: + truncate_oldest_actions(history, max_tokens=max_tokens, keep_n=keep_n) + + return _compact diff --git a/tests/adapters/test_history.py b/tests/adapters/test_history.py new file mode 100644 index 0000000000..3e7594c9a3 --- /dev/null +++ b/tests/adapters/test_history.py @@ -0,0 +1,93 @@ +from dspy.adapters.types.history import ( + History, + HistoryFrame, + Observation, + make_truncate_oldest_actions, + truncate_oldest_actions, +) + + +def test_legacy_messages_key_still_constructs_history_frames(): + legacy_message = {"question": "What is the capital of France?", "answer": "Paris"} + + history = History(messages=[legacy_message]) + + assert history.frames[0] == legacy_message + assert history.messages is history.frames + assert history.model_dump() == {"frames": [legacy_message]} + + +def test_field_frames_round_trip(): + history = History(frames=[]) + + history.append_inputs({"question": "hi"}) + history.append_outputs( + {"next_thought": "search first"}, + observations=[Observation(value="result", source="tool", call_id="call_0", name="search")], + ) + history.append_output({"answer": "bye"}) + + assert isinstance(history.frames[0], HistoryFrame) + assert history.frames[0].inputs == {"question": "hi"} + assert history.frames[1].outputs == {"next_thought": "search first"} + assert history.frames[1].observations[0].call_id == "call_0" + assert history.frames[2].outputs == {"answer": "bye"} + assert history.frames[2].complete + assert History.model_validate(history.model_dump()) == history + + +def test_has_open_episode_tracks_input_and_complete_boundaries(): + history = History(messages=[{"question": "legacy", "answer": "legacy answer"}]) + + assert not history.has_open_episode() + + history.append_inputs({"question": "new"}) + assert history.has_open_episode() + + history.append_outputs({"next_thought": "working"}) + history.frames.append({"question": "another legacy message"}) + assert history.has_open_episode() + + history.append_output({"answer": "done"}) + history.frames.append({"question": "final legacy message"}) + assert not history.has_open_episode() + + +def test_compact_if_needed_calls_compact_fn_with_history(): + calls = [] + history = History(frames=[], compact_fn=calls.append) + + history.compact_if_needed() + + assert calls == [history] + + +def test_truncate_oldest_actions_keeps_recent_observed_frames_and_non_actions(): + legacy_message = {"question": "legacy", "answer": "legacy answer"} + history = History(frames=[HistoryFrame(inputs={"question": "new"}), legacy_message]) + + for index in range(5): + history.append_outputs( + {"next_thought": str(index)}, + observations=[Observation(value=f"result {index}", call_id=f"call_{index}")], + ) + history.append_output({"answer": "done"}) + + truncate_oldest_actions(history, max_tokens=0, keep_n=2) + + observed_frames = [frame for frame in history.frames if isinstance(frame, HistoryFrame) and frame.observations] + assert [frame.outputs["next_thought"] for frame in observed_frames] == ["3", "4"] + assert history.frames[0] == HistoryFrame(inputs={"question": "new"}) + assert history.frames[1] == legacy_message + assert history.frames[-1] == HistoryFrame(outputs={"answer": "done"}, complete=True) + + +def test_make_truncate_oldest_actions_returns_compaction_fn(): + history = History(frames=[]) + for index in range(4): + history.append_outputs({"next_thought": str(index)}, observations=[Observation(value=f"result {index}")]) + + make_truncate_oldest_actions(max_tokens=0, keep_n=1)(history) + + observed_frames = [frame for frame in history.frames if isinstance(frame, HistoryFrame) and frame.observations] + assert [frame.outputs["next_thought"] for frame in observed_frames] == ["3"]