From 4bc80ca32bfb6b02d2b9577d7a047d09a02ef973 Mon Sep 17 00:00:00 2001 From: Eugene Eisenstein Date: Thu, 3 Sep 2026 13:19:35 -0400 Subject: [PATCH] feat(sdk): surface dialectic evidence in the Python SDK Accept `include_evidence` on peer and workspace chat, sync and async. Opting in returns a `ChatResponse` carrying the answer alongside what the dialectic read to produce it; leaving it out returns the answer on its own, so existing callers are unaffected. Overloads discriminate on the flag's literal value, so a Pydantic `response_format` combined with evidence types as `ChatResponse[Model]` while the default call still types as `str | None`. The four chat methods shared identical response-reading logic, which now lives in one `parse_chat_response` so they cannot drift. Evidence for a streamed answer can only be known once the answer is complete, so the server sends it on the stream's terminal event. `SSEStreamParser` now keeps it instead of discarding that event, and the stream response exposes it as `evidence` once drained. Reading it mid-stream returns None. Co-Authored-By: Claude Opus 5 (1M context) --- sdks/python/src/honcho/__init__.py | 10 + sdks/python/src/honcho/aio.py | 108 +++++++--- sdks/python/src/honcho/api_types.py | 45 +++++ sdks/python/src/honcho/client.py | 44 +++-- sdks/python/src/honcho/peer.py | 119 ++++++++++-- sdks/python/src/honcho/types.py | 67 ++++++- sdks/python/src/honcho/utils/sse.py | 38 +++- tests/sdk/test_chat_evidence.py | 292 ++++++++++++++++++++++++++++ 8 files changed, 665 insertions(+), 58 deletions(-) create mode 100644 tests/sdk/test_chat_evidence.py diff --git a/sdks/python/src/honcho/__init__.py b/sdks/python/src/honcho/__init__.py index a07bf35f..99903943 100644 --- a/sdks/python/src/honcho/__init__.py +++ b/sdks/python/src/honcho/__init__.py @@ -43,6 +43,10 @@ import re from .aio import ConclusionsViewAio, HonchoAio, PeerAio, ScopeAio, SessionAio from .api_types import ( + Evidence, + EvidenceMessageRef, + EvidenceObservation, + EvidenceToolCall, MessageCreateParams, ScopeBackfillJob, ScopeResponse, @@ -73,6 +77,7 @@ from .session import Session from .session_context import SessionContext, SessionSummaries, Summary from .types import ( AsyncDialecticStreamResponse, + ChatResponse, DialecticStreamResponse, ) @@ -138,7 +143,12 @@ __all__ = [ "SyncPage", # Streaming "AsyncDialecticStreamResponse", + "ChatResponse", "DialecticStreamResponse", + "Evidence", + "EvidenceMessageRef", + "EvidenceObservation", + "EvidenceToolCall", # Exceptions "APIError", "AuthenticationError", diff --git a/sdks/python/src/honcho/aio.py b/sdks/python/src/honcho/aio.py index 3629551f..adefeee0 100644 --- a/sdks/python/src/honcho/aio.py +++ b/sdks/python/src/honcho/aio.py @@ -60,8 +60,9 @@ from .message import Message from .mixins import AsyncMetadataConfigMixin from .pagination import AsyncPage from .session_context import SessionContext, SessionSummaries, Summary -from .types import AsyncDialecticStreamResponse +from .types import AsyncDialecticStreamResponse, ChatResponse from .utils import ( + SSEStreamParser, datetime_to_iso, normalize_peers_to_dict, parse_sse_astream, @@ -79,7 +80,12 @@ if TYPE_CHECKING: from .conclusions import ConclusionsView from .conclusions import ConclusionCreateParams -from .peer import Peer, TResponseFormat, serialize_response_format +from .peer import ( + Peer, + TResponseFormat, + parse_chat_response, + serialize_response_format, +) from .scope import Scope from .session import Session @@ -509,7 +515,8 @@ class HonchoAio(AsyncMetadataConfigMixin): | None = None, response_format: type[BaseModel] | dict[str, Any] | None = None, scope: str | list[str] | None = None, - ) -> BaseModel | str | None: + include_evidence: bool = False, + ) -> ChatResponse[Any] | BaseModel | str | None: """Query the entire workspace asynchronously (see Honcho.chat).""" await self._honcho._ensure_workspace_async() resolved_session_id = resolve_id(session) @@ -523,17 +530,14 @@ class HonchoAio(AsyncMetadataConfigMixin): response_format_schema = serialize_response_format(response_format) if response_format_schema is not None: body["response_format"] = response_format_schema + if include_evidence: + body["include_evidence"] = True data = await self._honcho._async_http_client.post( routes.workspace_chat(self._honcho.workspace_id), body=body, ) - content = data.get("content") - if not content: - return None - if isinstance(response_format, type): - return response_format.model_validate_json(content) - return content + return parse_chat_response(data, response_format, include_evidence) @validate_call(config=ConfigDict(arbitrary_types_allowed=True)) async def chat_stream( @@ -545,8 +549,13 @@ class HonchoAio(AsyncMetadataConfigMixin): | None = None, response_format: type[BaseModel] | dict[str, Any] | None = None, scope: str | list[str] | None = None, + include_evidence: bool = False, ) -> AsyncDialecticStreamResponse: - """Streaming variant of :meth:`chat` (async).""" + """Streaming variant of :meth:`chat` (async). + + With include_evidence, the returned stream's `evidence` is populated + once it has been fully consumed. + """ await self._honcho._ensure_workspace_async() resolved_session_id = resolve_id(session) body: dict[str, Any] = {"query": query, "stream": True} @@ -559,6 +568,12 @@ class HonchoAio(AsyncMetadataConfigMixin): response_format_schema = serialize_response_format(response_format) if response_format_schema is not None: body["response_format"] = response_format_schema + if include_evidence: + body["include_evidence"] = True + + # The parser holds the evidence that arrives on the final event, so it + # has to outlive the generator that drains the stream. + parser = SSEStreamParser() async def stream_response() -> AsyncGenerator[str, None]: async for chunk in parse_sse_astream( @@ -566,11 +581,12 @@ class HonchoAio(AsyncMetadataConfigMixin): "POST", routes.workspace_chat(self._honcho.workspace_id), body=body, - ) + ), + parser=parser, ): yield chunk - return AsyncDialecticStreamResponse(stream_response()) + return AsyncDialecticStreamResponse(stream_response(), lambda: parser.evidence) @validate_call async def search( @@ -777,9 +793,26 @@ class PeerAio(AsyncMetadataConfigMixin): reasoning_level: Literal["minimal", "low", "medium", "high", "max"] | None = None, response_format: type[TResponseFormat], + include_evidence: Literal[False] = False, timeout: float | None = None, ) -> TResponseFormat | None: ... + @overload + async def chat( + self, + query: str, + *, + target: str | PeerBase | None = None, + session: str | SessionBase | None = None, + scope: str | ScopeBase | Sequence[str | ScopeBase] | None = None, + sessions: Sequence[str | SessionBase] | None = None, + reasoning_level: Literal["minimal", "low", "medium", "high", "max"] + | None = None, + response_format: type[TResponseFormat], + include_evidence: Literal[True], + timeout: float | None = None, + ) -> ChatResponse[TResponseFormat]: ... + @overload async def chat( self, @@ -792,6 +825,23 @@ class PeerAio(AsyncMetadataConfigMixin): reasoning_level: Literal["minimal", "low", "medium", "high", "max"] | None = None, response_format: dict[str, Any] | None = None, + include_evidence: Literal[True], + timeout: float | None = None, + ) -> ChatResponse[str]: ... + + @overload + async def chat( + self, + query: str, + *, + target: str | PeerBase | None = None, + session: str | SessionBase | None = None, + scope: str | ScopeBase | Sequence[str | ScopeBase] | None = None, + sessions: Sequence[str | SessionBase] | None = None, + reasoning_level: Literal["minimal", "low", "medium", "high", "max"] + | None = None, + response_format: dict[str, Any] | None = None, + include_evidence: Literal[False] = False, timeout: float | None = None, ) -> str | None: ... @@ -807,17 +857,19 @@ class PeerAio(AsyncMetadataConfigMixin): reasoning_level: Literal["minimal", "low", "medium", "high", "max"] | None = None, response_format: type[BaseModel] | dict[str, Any] | None = None, + include_evidence: bool = False, timeout: float | None = Field( None, gt=0, description="Timeout in seconds for this chat request" ), - ) -> BaseModel | str | None: + ) -> ChatResponse[Any] | BaseModel | str | None: """Query the peer's representation asynchronously. See Peer.chat for parameter details. When response_format is a Pydantic model class, the answer is parsed into an instance of it; when it is a - JSON Schema dict, the answer is a JSON string. When timeout is omitted, - the Honcho client's configured timeout is used; retries can extend total - elapsed time. + JSON Schema dict, the answer is a JSON string. With include_evidence, + the answer comes back in a ChatResponse alongside what the dialectic + read to produce it. When timeout is omitted, the Honcho client's + configured timeout is used; retries can extend total elapsed time. """ await self._peer._honcho._ensure_workspace_async() target_id = resolve_id(target) @@ -838,18 +890,15 @@ class PeerAio(AsyncMetadataConfigMixin): response_format_schema = serialize_response_format(response_format) if response_format_schema is not None: body["response_format"] = response_format_schema + if include_evidence: + body["include_evidence"] = True data = await self._peer._honcho._async_http_client.post( routes.peer_chat(self._peer.workspace_id, self._peer.id), body=body, timeout=timeout, ) - content = data.get("content") - if not content: - return None - if isinstance(response_format, type): - return response_format.model_validate_json(content) - return content + return parse_chat_response(data, response_format, include_evidence) @validate_call(config=ConfigDict(arbitrary_types_allowed=True)) async def chat_stream( @@ -863,12 +912,14 @@ class PeerAio(AsyncMetadataConfigMixin): reasoning_level: Literal["minimal", "low", "medium", "high", "max"] | None = None, response_format: type[BaseModel] | dict[str, Any] | None = None, + include_evidence: bool = False, ) -> AsyncDialecticStreamResponse: """Query the peer's representation with streaming asynchronously. See Peer.chat_stream for parameter details. With response_format set, chunks stay raw text that accumulates to a JSON string; parse it after - the stream completes. + the stream completes. With include_evidence, the returned stream's + `evidence` is populated once it has been fully consumed. """ await self._peer._honcho._ensure_workspace_async() target_id = resolve_id(target) @@ -889,6 +940,12 @@ class PeerAio(AsyncMetadataConfigMixin): response_format_schema = serialize_response_format(response_format) if response_format_schema is not None: body["response_format"] = response_format_schema + if include_evidence: + body["include_evidence"] = True + + # The parser holds the evidence that arrives on the final event, so it + # has to outlive the generator that drains the stream. + parser = SSEStreamParser() async def stream_response() -> AsyncGenerator[str, None]: async for content in parse_sse_astream( @@ -896,11 +953,12 @@ class PeerAio(AsyncMetadataConfigMixin): "POST", routes.peer_chat(self._peer.workspace_id, self._peer.id), body=body, - ) + ), + parser=parser, ): yield content - return AsyncDialecticStreamResponse(stream_response()) + return AsyncDialecticStreamResponse(stream_response(), lambda: parser.evidence) async def sessions( self, diff --git a/sdks/python/src/honcho/api_types.py b/sdks/python/src/honcho/api_types.py index 23692f2f..ac64003c 100644 --- a/sdks/python/src/honcho/api_types.py +++ b/sdks/python/src/honcho/api_types.py @@ -543,12 +543,56 @@ class DialecticParams(BaseModel): stream: bool = False reasoning_level: ReasoningLevel = "low" response_format: dict[str, Any] | None = None + include_evidence: bool = False + + +class EvidenceObservation(BaseModel): + """A conclusion the dialectic read while answering.""" + + id: str + level: ConclusionLevel + content: str + created_at: datetime.datetime + session_id: str | None = None + source_ids: list[str] = Field(default_factory=list) + + +class EvidenceMessageRef(BaseModel): + """A message the dialectic read while answering.""" + + id: str + session_id: str + peer_id: str + content_preview: str + created_at: datetime.datetime + + +class EvidenceToolCall(BaseModel): + """A tool the dialectic invoked while answering.""" + + tool_name: str + tool_input: dict[str, Any] = Field(default_factory=dict) + + +class Evidence(BaseModel): + """What the dialectic read and did while answering. + + Collated from what the agent accessed rather than reported by the model, so + it over-reports: a listed conclusion was read, which is not proof the + answer leaned on it. `tool_calls` omits results and failed calls. + """ + + conclusions: list[EvidenceObservation] = Field(default_factory=list) + messages: list[EvidenceMessageRef] = Field(default_factory=list) + tool_calls: list[EvidenceToolCall] = Field(default_factory=list) + reasoning_trace_id: str | None = None class DialecticResponse(BaseModel): """Dialectic chat API response.""" content: str | None + evidence: Evidence | None = None class DialecticStreamDelta(BaseModel): @@ -562,6 +606,7 @@ class DialecticStreamChunk(BaseModel): delta: DialecticStreamDelta done: bool = False + evidence: Evidence | None = None # ============================================================================== diff --git a/sdks/python/src/honcho/client.py b/sdks/python/src/honcho/client.py index dbee9478..1bf28f30 100644 --- a/sdks/python/src/honcho/client.py +++ b/sdks/python/src/honcho/client.py @@ -28,11 +28,12 @@ from .http import AsyncHonchoHTTPClient, HonchoHTTPClient, routes from .message import Message from .mixins import MetadataConfigMixin from .pagination import SyncPage -from .peer import Peer, serialize_response_format +from .peer import Peer, parse_chat_response, serialize_response_format from .scope import Scope from .session import Session -from .types import DialecticStreamResponse +from .types import ChatResponse, DialecticStreamResponse from .utils import ( + SSEStreamParser, normalize_peers_to_dict, parse_sse_stream, resolve_id, @@ -702,7 +703,8 @@ class Honcho(BaseModel, MetadataConfigMixin): # pyright: ignore[reportUnsafeMul | None = None, response_format: type[BaseModel] | dict[str, Any] | None = None, scope: str | list[str] | None = None, - ) -> BaseModel | str | None: + include_evidence: bool = False, + ) -> ChatResponse[Any] | BaseModel | str | None: """ Query the entire workspace with a natural language question. @@ -720,9 +722,16 @@ class Honcho(BaseModel, MetadataConfigMixin): # pyright: ignore[reportUnsafeMul JSON Schema dict (returns a JSON string). scope: Optional scope name(s) restricting recall to those scopes' member sessions. Mutually exclusive with `session`. + include_evidence: When True, returns a `ChatResponse` carrying the + answer alongside what the dialectic read to produce it. + Evidence is collated from the agent's own reads rather than + reported by the model, so it is broader than a citation + list. Returns: - The synthesized answer, or None if no relevant information. + The synthesized answer, or None if no relevant information. With + `include_evidence=True`, a `ChatResponse` wrapping that same + content plus its evidence. """ self._ensure_workspace() resolved_session_id = resolve_id(session) @@ -736,17 +745,14 @@ class Honcho(BaseModel, MetadataConfigMixin): # pyright: ignore[reportUnsafeMul response_format_schema = serialize_response_format(response_format) if response_format_schema is not None: body["response_format"] = response_format_schema + if include_evidence: + body["include_evidence"] = True data = self._http.post( routes.workspace_chat(self.workspace_id), body=body, ) - content = data.get("content") - if not content: - return None - if isinstance(response_format, type): - return response_format.model_validate_json(content) - return content + return parse_chat_response(data, response_format, include_evidence) @validate_call(config=ConfigDict(arbitrary_types_allowed=True)) def chat_stream( @@ -758,8 +764,13 @@ class Honcho(BaseModel, MetadataConfigMixin): # pyright: ignore[reportUnsafeMul | None = None, response_format: type[BaseModel] | dict[str, Any] | None = None, scope: str | list[str] | None = None, + include_evidence: bool = False, ) -> DialecticStreamResponse: - """Streaming variant of :meth:`chat`. See chat() for argument docs.""" + """Streaming variant of :meth:`chat`. See chat() for argument docs. + + With include_evidence, the returned stream's `evidence` is populated + once it has been fully consumed. + """ self._ensure_workspace() resolved_session_id = resolve_id(session) body: dict[str, Any] = {"query": query, "stream": True} @@ -772,6 +783,12 @@ class Honcho(BaseModel, MetadataConfigMixin): # pyright: ignore[reportUnsafeMul response_format_schema = serialize_response_format(response_format) if response_format_schema is not None: body["response_format"] = response_format_schema + if include_evidence: + body["include_evidence"] = True + + # The parser holds the evidence that arrives on the final event, so it + # has to outlive the generator that drains the stream. + parser = SSEStreamParser() def stream_response() -> Generator[str, None, None]: yield from parse_sse_stream( @@ -779,10 +796,11 @@ class Honcho(BaseModel, MetadataConfigMixin): # pyright: ignore[reportUnsafeMul "POST", routes.workspace_chat(self.workspace_id), body=body, - ) + ), + parser=parser, ) - return DialecticStreamResponse(stream_response()) + return DialecticStreamResponse(stream_response(), lambda: parser.evidence) @validate_call def search( diff --git a/sdks/python/src/honcho/peer.py b/sdks/python/src/honcho/peer.py index bf6cf2d7..b564334f 100644 --- a/sdks/python/src/honcho/peer.py +++ b/sdks/python/src/honcho/peer.py @@ -12,6 +12,7 @@ from typing import TYPE_CHECKING, Any, Literal, TypeVar, overload from pydantic import BaseModel, ConfigDict, Field, PrivateAttr, validate_call from .api_types import ( + Evidence, MessageCreateParams, MessageResponse, PeerCardResponse, @@ -28,8 +29,14 @@ from .http import routes from .message import Message from .mixins import MetadataConfigMixin from .pagination import SyncPage -from .types import DialecticStreamResponse -from .utils import parse_datetime, parse_sse_stream, resolve_id, scope_recall_fields +from .types import ChatResponse, DialecticStreamResponse +from .utils import ( + SSEStreamParser, + parse_datetime, + parse_sse_stream, + resolve_id, + scope_recall_fields, +) if TYPE_CHECKING: from .aio import PeerAio @@ -52,6 +59,41 @@ def serialize_response_format( return response_format +def parse_chat_response( + data: dict[str, Any], + response_format: type[BaseModel] | dict[str, Any] | None, + include_evidence: bool, +) -> ChatResponse[Any] | BaseModel | str | None: + """Read a chat response body into whatever the caller asked for. + + Shared by peer and workspace chat, sync and async, so the four stay in + step. Without evidence the answer is returned bare; with it, the answer and + its evidence come back together. An empty answer + stays falsy either way -- as `None`, or as a `ChatResponse` whose content + is None -- so evidence is still available for a run that found nothing to + say. + """ + content = data.get("content") + parsed: BaseModel | str | None = None + if content: + parsed = ( + response_format.model_validate_json(content) + if isinstance(response_format, type) + else content + ) + + if not include_evidence: + return parsed + + raw_evidence = data.get("evidence") + return ChatResponse( + content=parsed, + evidence=Evidence.model_validate(raw_evidence) + if isinstance(raw_evidence, dict) + else None, + ) + + class Peer(PeerBase, MetadataConfigMixin): """ Represents a peer in the Honcho system. @@ -246,9 +288,26 @@ class Peer(PeerBase, MetadataConfigMixin): reasoning_level: Literal["minimal", "low", "medium", "high", "max"] | None = None, response_format: type[TResponseFormat], + include_evidence: Literal[False] = False, timeout: float | None = None, ) -> TResponseFormat | None: ... + @overload + def chat( + self, + query: str, + *, + target: str | PeerBase | None = None, + session: str | SessionBase | None = None, + scope: str | ScopeBase | Sequence[str | ScopeBase] | None = None, + sessions: Sequence[str | SessionBase] | None = None, + reasoning_level: Literal["minimal", "low", "medium", "high", "max"] + | None = None, + response_format: type[TResponseFormat], + include_evidence: Literal[True], + timeout: float | None = None, + ) -> ChatResponse[TResponseFormat]: ... + @overload def chat( self, @@ -261,6 +320,23 @@ class Peer(PeerBase, MetadataConfigMixin): reasoning_level: Literal["minimal", "low", "medium", "high", "max"] | None = None, response_format: dict[str, Any] | None = None, + include_evidence: Literal[True], + timeout: float | None = None, + ) -> ChatResponse[str]: ... + + @overload + def chat( + self, + query: str, + *, + target: str | PeerBase | None = None, + session: str | SessionBase | None = None, + scope: str | ScopeBase | Sequence[str | ScopeBase] | None = None, + sessions: Sequence[str | SessionBase] | None = None, + reasoning_level: Literal["minimal", "low", "medium", "high", "max"] + | None = None, + response_format: dict[str, Any] | None = None, + include_evidence: Literal[False] = False, timeout: float | None = None, ) -> str | None: ... @@ -276,10 +352,11 @@ class Peer(PeerBase, MetadataConfigMixin): reasoning_level: Literal["minimal", "low", "medium", "high", "max"] | None = None, response_format: type[BaseModel] | dict[str, Any] | None = None, + include_evidence: bool = False, timeout: float | None = Field( None, gt=0, description="Timeout in seconds for this chat request" ), - ) -> BaseModel | str | None: + ) -> ChatResponse[Any] | BaseModel | str | None: """ Query the peer's representation with a natural language question. @@ -315,6 +392,12 @@ class Peer(PeerBase, MetadataConfigMixin): model class to get a parsed instance back, or a raw JSON Schema dict (root type "object") to get the answer as a JSON string. + include_evidence: When True, returns a ``ChatResponse`` carrying the + answer alongside what the dialectic read to produce it. + Evidence is collated from the agent's own reads rather + than reported by the model, so it is broader than a + citation list: a conclusion appears because the agent saw + it, not as proof the answer relied on it. timeout: Optional timeout in seconds for each HTTP attempt made by this request. When omitted, the Honcho client's configured timeout is used. Retries can extend total elapsed time. @@ -322,7 +405,9 @@ class Peer(PeerBase, MetadataConfigMixin): Returns: Response string containing the answer (a JSON string when a schema dict was given), a parsed model instance when a Pydantic model class - was given, or None if no relevant information. + was given, or None if no relevant information. With + ``include_evidence=True``, a ``ChatResponse`` wrapping that same + content plus its evidence. Raises: ValueError: If ``scope`` is combined with ``session`` or ``sessions``. @@ -346,18 +431,15 @@ class Peer(PeerBase, MetadataConfigMixin): response_format_schema = serialize_response_format(response_format) if response_format_schema is not None: body["response_format"] = response_format_schema + if include_evidence: + body["include_evidence"] = True data = self._honcho._http.post( routes.peer_chat(self.workspace_id, self.id), body=body, timeout=timeout, ) - content = data.get("content") - if not content: - return None - if isinstance(response_format, type): - return response_format.model_validate_json(content) - return content + return parse_chat_response(data, response_format, include_evidence) @validate_call(config=ConfigDict(arbitrary_types_allowed=True)) def chat_stream( @@ -371,6 +453,7 @@ class Peer(PeerBase, MetadataConfigMixin): reasoning_level: Literal["minimal", "low", "medium", "high", "max"] | None = None, response_format: type[BaseModel] | dict[str, Any] | None = None, + include_evidence: bool = False, ) -> DialecticStreamResponse: """ Query the peer's representation with a natural language question, streaming the response. @@ -398,6 +481,11 @@ class Peer(PeerBase, MetadataConfigMixin): Streamed chunks stay raw text that accumulates to a JSON string; parse it yourself (e.g. with Model.model_validate_json) once the stream completes. + include_evidence: When True, the returned stream's ``evidence`` + attribute is populated once the stream has been + fully consumed. Evidence cannot be known before + the answer is complete, so the server sends it on + the stream's final event. Returns: DialecticStreamResponse object that can be iterated over and provides final response @@ -424,6 +512,12 @@ class Peer(PeerBase, MetadataConfigMixin): response_format_schema = serialize_response_format(response_format) if response_format_schema is not None: body["response_format"] = response_format_schema + if include_evidence: + body["include_evidence"] = True + + # The parser holds the evidence that arrives on the final event, so it + # has to outlive the generator that drains the stream. + parser = SSEStreamParser() def stream_response() -> Generator[str, None, None]: yield from parse_sse_stream( @@ -431,10 +525,11 @@ class Peer(PeerBase, MetadataConfigMixin): "POST", routes.peer_chat(self.workspace_id, self.id), body=body, - ) + ), + parser=parser, ) - return DialecticStreamResponse(stream_response()) + return DialecticStreamResponse(stream_response(), lambda: parser.evidence) def sessions( self, diff --git a/sdks/python/src/honcho/types.py b/sdks/python/src/honcho/types.py index 5d02be18..dacf6036 100644 --- a/sdks/python/src/honcho/types.py +++ b/sdks/python/src/honcho/types.py @@ -2,14 +2,38 @@ from __future__ import annotations -from collections.abc import AsyncIterator, Iterator -from typing import Self +from collections.abc import AsyncIterator, Callable, Iterator +from dataclasses import dataclass +from typing import Generic, Self, TypeVar + +from honcho.api_types import Evidence __all__ = [ + "ChatResponse", "DialecticStreamResponse", "AsyncDialecticStreamResponse", ] +TContent = TypeVar("TContent") + + +@dataclass(frozen=True) +class ChatResponse(Generic[TContent]): + """An answer together with what it was built from. + + Returned by `chat(..., include_evidence=True)`. Without that, `chat` + returns the answer on its own. + + `evidence` lists what the dialectic read while answering, collated from its + own reads rather than reported by the model. That makes it deterministic + but broader than a citation list: a conclusion appears because the agent + saw it, which is not proof the answer relied on it. It is empty rather than + None when a run genuinely read nothing. + """ + + content: TContent | None + evidence: Evidence | None = None + class DialecticStreamResponse: """ @@ -35,11 +59,17 @@ class DialecticStreamResponse: _iterator: Iterator[str] _accumulated_content: list[str] _is_complete: bool + _evidence_source: Callable[[], Evidence | None] | None - def __init__(self, iterator: Iterator[str]) -> None: + def __init__( + self, + iterator: Iterator[str], + evidence_source: Callable[[], Evidence | None] | None = None, + ) -> None: self._iterator = iterator self._accumulated_content = [] self._is_complete = False + self._evidence_source = evidence_source def __iter__(self) -> Self: return self @@ -66,6 +96,19 @@ class DialecticStreamResponse: """ return {"content": "".join(self._accumulated_content)} + @property + def evidence(self) -> Evidence | None: + """What the answer was built from, once the stream has finished. + + The server can only know this after the answer is complete, so it + arrives on the stream's final event. Reading it before the stream is + fully consumed returns None, as does a request that did not ask for + evidence. + """ + if self._evidence_source is None: + return None + return self._evidence_source() + @property def is_complete(self) -> bool: """Check if the stream has finished.""" @@ -96,11 +139,17 @@ class AsyncDialecticStreamResponse: _iterator: AsyncIterator[str] _accumulated_content: list[str] _is_complete: bool + _evidence_source: Callable[[], Evidence | None] | None - def __init__(self, iterator: AsyncIterator[str]) -> None: + def __init__( + self, + iterator: AsyncIterator[str], + evidence_source: Callable[[], Evidence | None] | None = None, + ) -> None: self._iterator = iterator self._accumulated_content = [] self._is_complete = False + self._evidence_source = evidence_source def __aiter__(self) -> Self: return self @@ -127,6 +176,16 @@ class AsyncDialecticStreamResponse: """ return {"content": "".join(self._accumulated_content)} + @property + def evidence(self) -> Evidence | None: + """What the answer was built from, once the stream has finished. + + See :attr:`DialecticStreamResponse.evidence`. + """ + if self._evidence_source is None: + return None + return self._evidence_source() + @property def is_complete(self) -> bool: """Check if the stream has finished.""" diff --git a/sdks/python/src/honcho/utils/sse.py b/sdks/python/src/honcho/utils/sse.py index 5c24ea64..5603a284 100644 --- a/sdks/python/src/honcho/utils/sse.py +++ b/sdks/python/src/honcho/utils/sse.py @@ -8,6 +8,10 @@ import logging from collections.abc import AsyncGenerator, AsyncIterable, Generator, Iterable from typing import Any, cast +from pydantic import ValidationError + +from honcho.api_types import Evidence + logger = logging.getLogger(__name__) @@ -26,6 +30,7 @@ class SSEStreamParser: - `done: true` to indicate stream completion. - `delta.content` containing incremental text. + - `evidence` on the terminal message, when the request asked for it. Any JSON decoding failures are logged with the same warning format as the legacy parser, including a preview of the data payload. @@ -37,12 +42,23 @@ class SSEStreamParser: )(errors="replace") self._text_buffer: str = "" self._done: bool = False + self._evidence: Evidence | None = None @property def done(self) -> bool: """Whether the stream has emitted a `done: true` message.""" return self._done + @property + def evidence(self) -> Evidence | None: + """What the answer was built from, once the stream has finished. + + Evidence can only be known after the answer is complete, so the server + sends it on the terminal message. It stays None until then, and stays + None throughout unless the request asked for it. + """ + return self._evidence + def feed(self, chunk: bytes) -> Generator[str, None, None]: """ Feed the next bytes from the SSE stream and yield any newly available content. @@ -145,6 +161,12 @@ class SSEStreamParser: chunk_data = cast(dict[str, Any], parsed) if chunk_data.get("done"): self._done = True + evidence = chunk_data.get("evidence") + if isinstance(evidence, dict): + try: + self._evidence = Evidence.model_validate(evidence) + except ValidationError as e: + logger.warning("Failed to decode streamed evidence: %s", e) return delta_obj = chunk_data.get("delta", {}) @@ -190,17 +212,21 @@ def parse_sse_chunk( yield from parser.feed(chunk) -def parse_sse_stream(chunks: Iterable[bytes]) -> Generator[str, None, None]: +def parse_sse_stream( + chunks: Iterable[bytes], *, parser: SSEStreamParser | None = None +) -> Generator[str, None, None]: """ Parse an SSE byte stream and yield content strings. Args: chunks: An iterable of raw byte chunks from an SSE stream. + parser: Optional parser to use, for callers that need to read state off + it once the stream finishes -- `evidence`, say. Yields: Content strings extracted from delta objects, in order. """ - parser = SSEStreamParser() + parser = parser or SSEStreamParser() for chunk in chunks: yield from parser.feed(chunk) if parser.done: @@ -208,17 +234,21 @@ def parse_sse_stream(chunks: Iterable[bytes]) -> Generator[str, None, None]: yield from parser.finalize() -async def parse_sse_astream(chunks: AsyncIterable[bytes]) -> AsyncGenerator[str, None]: +async def parse_sse_astream( + chunks: AsyncIterable[bytes], *, parser: SSEStreamParser | None = None +) -> AsyncGenerator[str, None]: """ Parse an async SSE byte stream and yield content strings. Args: chunks: An async iterable of raw byte chunks from an SSE stream. + parser: Optional parser to use, for callers that need to read state off + it once the stream finishes -- `evidence`, say. Yields: Content strings extracted from delta objects, in order. """ - parser = SSEStreamParser() + parser = parser or SSEStreamParser() async for chunk in chunks: for content in parser.feed(chunk): yield content diff --git a/tests/sdk/test_chat_evidence.py b/tests/sdk/test_chat_evidence.py new file mode 100644 index 00000000..24121eb0 --- /dev/null +++ b/tests/sdk/test_chat_evidence.py @@ -0,0 +1,292 @@ +"""Tests for the SDK's `include_evidence` option on peer and workspace chat. + +The server's dialectic is mocked out by the autouse `mock_llm_call_functions` +fixture, so these are about the SDK contract: sending the flag, and giving the +caller back a typed answer-plus-evidence instead of a bare answer. What the +evidence actually contains is covered server-side. +""" + +from datetime import UTC, datetime +from typing import Any + +import pytest +from pydantic import BaseModel + +from sdks.python.src.honcho import ChatResponse, Evidence +from sdks.python.src.honcho.client import Honcho +from src import models + +TOOL_CALL = {"tool_name": "search_memory", "tool_input": {"query": "coffee"}} +QUERY = "What does the user drink?" +WORKSPACE_QUERY = "What do people here drink?" +NOW = datetime(2026, 1, 1, tzinfo=UTC) + + +def _server_document() -> models.Document: + """A conclusion row shaped the way the server would hand one over.""" + return models.Document( + id="doc-sentinel", + level="deductive", + content="User drinks coffee in the morning", + internal_metadata={}, + source_ids=["doc-a", "doc-b"], + session_name="session-1", + created_at=NOW, + observer="observer", + observed="observed", + workspace_name="workspace", + ) + + +def _server_message() -> models.Message: + return models.Message( + public_id="msg-sentinel", + content="I drink a lot of coffee", + peer_name="alice", + session_name="session-1", + created_at=NOW, + workspace_name="workspace", + seq_in_session=1, + ) + + +class Drink(BaseModel): + name: str + + +def _stub_chat(mock: Any, content: str) -> None: + """Have the mocked dialectic record evidence when it was handed a place to.""" + + async def _chat(*_args: object, **kwargs: Any) -> str: + evidence = kwargs.get("evidence") + if evidence is not None: + evidence.record_tool_calls([TOOL_CALL]) + return content + + mock.side_effect = _chat + + +def _stub_chat_stream(mock: Any) -> None: + def _chat_stream(*_args: object, **kwargs: Any) -> Any: + evidence = kwargs.get("evidence") + + async def _chunks() -> Any: + if evidence is not None: + evidence.record_tool_calls([TOOL_CALL]) + for chunk in ("The user ", "drinks coffee."): + yield chunk + + return _chunks() + + mock.side_effect = _chat_stream + + +class TestPeerChat: + def test_deserializes_the_evidence_payload( + self, + honcho_sync_test_client: Honcho, + mock_llm_call_functions: dict[str, Any], + ): + _stub_chat(mock_llm_call_functions["agentic_chat"], "The user drinks coffee.") + peer = honcho_sync_test_client.peer("alice") + + result = peer.chat(QUERY, include_evidence=True) + + assert isinstance(result, ChatResponse) + assert isinstance(result.evidence, Evidence) + assert result.evidence.tool_calls[0].tool_name == "search_memory" + + def test_sends_the_flag_only_when_asked( + self, + honcho_sync_test_client: Honcho, + mock_llm_call_functions: dict[str, Any], + ): + peer = honcho_sync_test_client.peer("alice") + + peer.chat(QUERY) + assert mock_llm_call_functions["agentic_chat"].await_args.kwargs[ + "evidence" + ] is (None) + + peer.chat(QUERY, include_evidence=True) + assert ( + mock_llm_call_functions["agentic_chat"].await_args.kwargs["evidence"] + is not None + ) + + def test_still_parses_a_response_format_alongside_evidence( + self, + honcho_sync_test_client: Honcho, + mock_llm_call_functions: dict[str, Any], + ): + """A schema and evidence are independent; asking for both works.""" + _stub_chat(mock_llm_call_functions["agentic_chat"], '{"name": "coffee"}') + peer = honcho_sync_test_client.peer("alice") + + result = peer.chat( + "What does the user drink?", + response_format=Drink, + include_evidence=True, + ) + + assert isinstance(result, ChatResponse) + assert isinstance(result.content, Drink) + assert result.content.name == "coffee" + assert result.evidence is not None + + def test_reports_evidence_even_when_there_is_no_answer( + self, + honcho_sync_test_client: Honcho, + mock_llm_call_functions: dict[str, Any], + ): + """An empty answer must not swallow the evidence alongside it.""" + _stub_chat(mock_llm_call_functions["agentic_chat"], "") + peer = honcho_sync_test_client.peer("alice") + + result = peer.chat(QUERY, include_evidence=True) + + assert isinstance(result, ChatResponse) + assert result.content is None + assert result.evidence is not None + assert result.evidence.tool_calls + + def test_deserializes_conclusions_and_messages_into_typed_objects( + self, + honcho_sync_test_client: Honcho, + mock_llm_call_functions: dict[str, Any], + ): + """The SDK has to parse the real wire shape, not just tool calls.""" + + async def _chat(*_args: object, **kwargs: Any) -> str: + evidence = kwargs.get("evidence") + if evidence is not None: + evidence.add_documents([_server_document()]) + evidence.add_messages([_server_message()]) + return "The user drinks coffee." + + mock_llm_call_functions["agentic_chat"].side_effect = _chat + peer = honcho_sync_test_client.peer("alice") + + result = peer.chat(QUERY, include_evidence=True) + + assert isinstance(result, ChatResponse) + assert result.evidence is not None + (conclusion,) = result.evidence.conclusions + assert conclusion.id == "doc-sentinel" + assert conclusion.level == "deductive" + assert conclusion.content == "User drinks coffee in the morning" + assert conclusion.source_ids == ["doc-a", "doc-b"] + assert conclusion.created_at.tzinfo is not None + (message,) = result.evidence.messages + assert message.id == "msg-sentinel" + assert message.peer_id == "alice" + assert message.created_at.tzinfo is not None + + +class TestPeerChatStream: + def test_evidence_stays_none_when_not_requested( + self, honcho_sync_test_client: Honcho + ): + peer = honcho_sync_test_client.peer("alice") + + stream = peer.chat_stream(QUERY) + list(stream) + + assert stream.evidence is None + + +class TestWorkspaceChat: + def test_returns_a_bare_answer_by_default(self, honcho_sync_test_client: Honcho): + answer = honcho_sync_test_client.chat(WORKSPACE_QUERY) + + assert isinstance(answer, str) + + def test_stream_evidence_is_available_once_it_drains( + self, + honcho_sync_test_client: Honcho, + mock_llm_call_functions: dict[str, Any], + ): + _stub_chat_stream(mock_llm_call_functions["workspace_chat_stream"]) + + stream = honcho_sync_test_client.chat_stream( + WORKSPACE_QUERY, include_evidence=True + ) + list(stream) + + assert stream.evidence is not None + + +@pytest.mark.asyncio +class TestSyncAndAsyncParity: + """The async accessors have to return the same shapes as the sync ones.""" + + async def test_peer_chat_returns_answer_and_evidence( + self, + client_fixture: tuple[Honcho, str], + mock_llm_call_functions: dict[str, Any], + ): + honcho, client_type = client_fixture + _stub_chat(mock_llm_call_functions["agentic_chat"], "The user drinks coffee.") + + if client_type == "async": + peer = await honcho.aio.peer("alice") + result = await peer.aio.chat(QUERY, include_evidence=True) + else: + result = honcho.peer("alice").chat(QUERY, include_evidence=True) + + assert isinstance(result, ChatResponse) + assert result.content == "The user drinks coffee." + assert result.evidence is not None + + async def test_peer_chat_returns_a_bare_answer_by_default( + self, client_fixture: tuple[Honcho, str] + ): + honcho, client_type = client_fixture + + if client_type == "async": + peer = await honcho.aio.peer("alice") + answer = await peer.aio.chat(QUERY) + else: + answer = honcho.peer("alice").chat(QUERY) + + assert isinstance(answer, str) + + async def test_peer_chat_stream_evidence_is_available_once_it_drains( + self, + client_fixture: tuple[Honcho, str], + mock_llm_call_functions: dict[str, Any], + ): + honcho, client_type = client_fixture + _stub_chat_stream(mock_llm_call_functions["agentic_chat_stream"]) + + if client_type == "async": + peer = await honcho.aio.peer("alice") + stream = await peer.aio.chat_stream(QUERY, include_evidence=True) + chunks = [chunk async for chunk in stream] + else: + sync_stream = honcho.peer("alice").chat_stream(QUERY, include_evidence=True) + chunks = list(sync_stream) + stream = sync_stream + + assert "".join(chunks) == "The user drinks coffee." + assert stream.evidence is not None + assert stream.evidence.tool_calls[0].tool_name == "search_memory" + + async def test_workspace_chat_returns_answer_and_evidence( + self, + client_fixture: tuple[Honcho, str], + mock_llm_call_functions: dict[str, Any], + ): + honcho, client_type = client_fixture + _stub_chat( + mock_llm_call_functions["workspace_chat"], "People here drink coffee." + ) + + if client_type == "async": + result = await honcho.aio.chat(WORKSPACE_QUERY, include_evidence=True) + else: + result = honcho.chat(WORKSPACE_QUERY, include_evidence=True) + + assert isinstance(result, ChatResponse) + assert result.content == "People here drink coffee." + assert result.evidence is not None