feat(history): add typed history frames

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

dspy/adapters/__init__.py

@@ -2,14 +2,28 @@
 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__ = [
     "Adapter",
     "ChatAdapter",
     "Type",
     "History",
+    "HistoryFrame",
+    "Observation",
     "Image",
     "Audio",
     "File",

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",
+]

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

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"]