From 3a2d33130d54b213a4ce78e3298794e4586612fd Mon Sep 17 00:00:00 2001 From: Krishna Bhatnagar Date: Thu, 1 Oct 2026 05:03:27 -0400 Subject: [PATCH] feat: capture streamed answers whole; customer ids; outcome() - Streams: each chunk reaches the agent untouched; a copy is kept and, when the stream ends, assembled into the answer a non-streamed call would have (OpenAI chunks, Anthropic events, sync and async, compiled or forwarded). A stream the agent stops reading is kept and marked stream_complete: false. - Customers: conversation(id, customer=...) or customer_id= on a call; captured as end_user, scrubbed like everything else. - outcome(conversation_id, label): resolved, escalated, unresolved, abandoned, reopened or complaint, sent in the background to /v1/outcome. Co-Authored-By: Claude Opus 5.5 --- README.md | 19 ++ src/agentcompile/__init__.py | 10 +- src/agentcompile/_assemble.py | 188 ++++++++++++++++ src/agentcompile/_capture.py | 39 +++- src/agentcompile/_conversation.py | 16 +- src/agentcompile/_core.py | 46 ++-- src/agentcompile/_outcome.py | 68 ++++++ tests/test_capture_complete.py | 342 ++++++++++++++++++++++++++++++ 8 files changed, 708 insertions(+), 20 deletions(-) create mode 100644 src/agentcompile/_assemble.py create mode 100644 src/agentcompile/_outcome.py create mode 100644 tests/test_capture_complete.py diff --git a/README.md b/README.md index f2efb47..0782103 100644 --- a/README.md +++ b/README.md @@ -67,6 +67,25 @@ pip install -e ".[dev]" pytest tests ``` +## Outcomes and customers + +Tell AgentCompile how a conversation ended, so it learns only from the ones that went well: + +```python +agentcompile.outcome(ticket.id, "resolved") # or escalated, unresolved, abandoned, reopened, complaint +``` + +Pass a stable customer id so repeat jobs are counted per customer (scrubbed like everything +else when capture is on): + +```python +with agentcompile.conversation(ticket.id, customer=ticket.customer_id): + run_agent() +``` + +Streamed answers are captured too: your agent reads the stream as always, and the whole answer +is captured once it ends. + ## Privacy With `capture=True`, personal data is scrubbed on your machine before anything is sent: diff --git a/src/agentcompile/__init__.py b/src/agentcompile/__init__.py index 79d8c0e..c989444 100644 --- a/src/agentcompile/__init__.py +++ b/src/agentcompile/__init__.py @@ -19,10 +19,13 @@ from ._conversation import conversation from ._core import LIVE, SHADOW, Proxy, Router, async_create, sync_create from ._decide import AsyncDecider, Decider, Settings +from ._outcome import OUTCOMES, outcome +from ._outcome import flush as _flush_outcomes +from ._outcome import remember as _remember from ._scrub import load_key from ._trail import OnEvent, Trail -__all__ = ["LIVE", "SHADOW", "Settings", "conversation", "flush", "wrap"] +__all__ = ["LIVE", "OUTCOMES", "SHADOW", "Settings", "conversation", "flush", "outcome", "wrap"] try: from ._version import __version__ except ImportError: # running from a source checkout that was never built @@ -64,6 +67,7 @@ def wrap( timeout=timeout, ) the_trail = Trail(trail, on_event) + _remember(settings) # outcome() sends with the client wrapped last if capture is None: capture = os.environ.get("AGENTCOMPILE_CAPTURE", "") in ("1", "true", "yes") # One sender per wrapped client; it always sends from its own thread (sync client). @@ -103,8 +107,10 @@ def wrap( def flush(timeout: float = 5.0) -> None: - """Send every captured call still queued (short scripts and tests; also runs at exit).""" + """Send every captured call and outcome still queued (short scripts and tests; also runs + at exit).""" flush_all(timeout) + _flush_outcomes(timeout) def _router( diff --git a/src/agentcompile/_assemble.py b/src/agentcompile/_assemble.py new file mode 100644 index 0000000..bd59d1e --- /dev/null +++ b/src/agentcompile/_assemble.py @@ -0,0 +1,188 @@ +"""Streamed answers, captured whole: each chunk passes to your agent untouched as it arrives, a +copy is kept, and when the stream ends the pieces are assembled into the same shape a +non-streamed answer has (a chat completion, or an Anthropic message), so AgentCompile reads +streamed and non-streamed calls the same way.""" + +from __future__ import annotations + +import json +from collections.abc import AsyncIterator, Iterator +from typing import Any, Callable + +from ._payload import jsonable + +OnDone = Callable[[list[Any], bool], None] # (chunks as JSON, the stream ran to its end) + + +def assemble(provider: str, chunks: list[Any]) -> dict[str, Any] | None: + try: + return _openai(chunks) if provider == "openai" else _anthropic(chunks) + except Exception: # never break the agent over a shape we didn't expect + return None + + +def _openai(chunks: list[dict[str, Any]]) -> dict[str, Any]: + first = chunks[0] if chunks else {} + content: list[str] = [] + calls: dict[int, dict[str, Any]] = {} + finish = None + usage = None + for chunk in chunks: + usage = chunk.get("usage") or usage + for choice in chunk.get("choices") or []: + if choice.get("index", 0) != 0: + continue + delta = choice.get("delta") or {} + if delta.get("content"): + content.append(delta["content"]) + for call in delta.get("tool_calls") or []: + slot = calls.setdefault( + call.get("index", 0), + {"id": None, "type": "function", "function": {"name": "", "arguments": ""}}, + ) + slot["id"] = call.get("id") or slot["id"] + fn = call.get("function") or {} + slot["function"]["name"] += fn.get("name") or "" + slot["function"]["arguments"] += fn.get("arguments") or "" + finish = choice.get("finish_reason") or finish + message: dict[str, Any] = {"role": "assistant", "content": "".join(content) or None} + if calls: + message["tool_calls"] = [calls[i] for i in sorted(calls)] + out: dict[str, Any] = { + "id": first.get("id"), + "object": "chat.completion", + "created": first.get("created"), + "model": first.get("model"), + "choices": [{"index": 0, "message": message, "finish_reason": finish}], + } + if usage: + out["usage"] = usage + return out + + +def _anthropic(events: list[dict[str, Any]]) -> dict[str, Any]: + message: dict[str, Any] = {} + blocks: dict[int, dict[str, Any]] = {} + partial: dict[int, list[str]] = {} + for event in events: + kind = event.get("type") + if kind == "message_start": + message = dict(event.get("message") or {}) + elif kind == "content_block_start": + blocks[event["index"]] = dict(event.get("content_block") or {}) + elif kind == "content_block_delta": + block = blocks.setdefault(event["index"], {"type": "text", "text": ""}) + delta = event.get("delta") or {} + if delta.get("type") == "text_delta": + block["text"] = block.get("text", "") + delta.get("text", "") + elif delta.get("type") == "input_json_delta": + partial.setdefault(event["index"], []).append(delta.get("partial_json", "")) + elif delta.get("type") == "thinking_delta": + block["thinking"] = block.get("thinking", "") + delta.get("thinking", "") + elif kind == "message_delta": + delta = event.get("delta") or {} + message["stop_reason"] = delta.get("stop_reason", message.get("stop_reason")) + message["stop_sequence"] = delta.get("stop_sequence") + usage = dict(message.get("usage") or {}) + usage.update(event.get("usage") or {}) + message["usage"] = usage + for index, pieces in partial.items(): + text = "".join(pieces) + blocks[index]["input"] = json.loads(text) if text else {} + message["content"] = [blocks[i] for i in sorted(blocks)] + return message + + +class CapturingStream: + """Your stream, unchanged, with a copy of each chunk kept for capture.""" + + def __init__(self, stream: Any, on_done: OnDone) -> None: + self._stream = stream + self._on_done = on_done + self._chunks: list[Any] = [] + self._finished = False + + def __iter__(self) -> Iterator[Any]: + complete = False + try: + for item in self._stream: + self._chunks.append(jsonable(item)) + yield item + complete = True + finally: + self._finish(complete=complete) + + def __enter__(self) -> CapturingStream: + enter = getattr(self._stream, "__enter__", None) + if enter is not None: + enter() + return self + + def __exit__(self, *exc: Any) -> Any: + self._finish(complete=False) + leave = getattr(self._stream, "__exit__", None) + return leave(*exc) if leave is not None else None + + def close(self) -> None: + self._finish(complete=False) + close = getattr(self._stream, "close", None) + if close is not None: + close() + + def _finish(self, *, complete: bool) -> None: + if not self._finished: + self._finished = True + self._on_done(self._chunks, complete) + + def __getattr__(self, name: str) -> Any: + return getattr(self._stream, name) + + +class AsyncCapturingStream: + """The async version.""" + + def __init__(self, stream: Any, on_done: OnDone) -> None: + self._stream = stream + self._on_done = on_done + self._chunks: list[Any] = [] + self._finished = False + + async def _iterate(self) -> AsyncIterator[Any]: + complete = False + try: + async for item in self._stream: + self._chunks.append(jsonable(item)) + yield item + complete = True + finally: + self._finish(complete=complete) + + def __aiter__(self) -> AsyncIterator[Any]: + return self._iterate() + + async def __aenter__(self) -> AsyncCapturingStream: + enter = getattr(self._stream, "__aenter__", None) + if enter is not None: + await enter() + return self + + async def __aexit__(self, *exc: Any) -> Any: + self._finish(complete=False) + leave = getattr(self._stream, "__aexit__", None) + return (await leave(*exc)) if leave is not None else None + + async def close(self) -> None: + self._finish(complete=False) + close = getattr(self._stream, "close", None) + if close is not None: + result = close() + if hasattr(result, "__await__"): + await result + + def _finish(self, *, complete: bool) -> None: + if not self._finished: + self._finished = True + self._on_done(self._chunks, complete) + + def __getattr__(self, name: str) -> Any: + return getattr(self._stream, name) diff --git a/src/agentcompile/_capture.py b/src/agentcompile/_capture.py index 182624c..4e6cde5 100644 --- a/src/agentcompile/_capture.py +++ b/src/agentcompile/_capture.py @@ -18,9 +18,10 @@ import httpx +from ._assemble import assemble from ._decide import CONVERSATION_HEADER, Settings, _request from ._payload import jsonable, payload -from ._scrub import scrub_call +from ._scrub import scrub_call, scrub_value MAX_QUEUE = 2000 BATCH = 50 @@ -63,9 +64,11 @@ def add( kwargs: dict[str, Any], response: Any, stream: bool = False, + streamed: dict[str, Any] | None = None, + customer: str | None = None, ) -> None: - """Queue one call. Never raises. A streamed answer is not captured yet: the record - keeps the request and says so.""" + """Queue one call. Never raises. `stream` without `streamed`: the answer wasn't kept + (the record says so); `streamed`: the answer was assembled from its stream.""" try: request, answer = payload(kwargs), None if stream else jsonable(response) if self.scrub_key is not None: @@ -78,10 +81,20 @@ def add( "request": request, "response": answer, } + if customer: + # The customer's id, scrubbed like everything else (an email becomes a token). + record["end_user"] = ( + scrub_value(customer, self.scrub_key, "customer") + if self.scrub_key is not None + else customer + ) if self.scrub_key is not None: record["scrubbed"] = True if stream: record["stream"] = True + if streamed is not None: + record["stream"] = True + record["stream_complete"] = streamed["complete"] with self._lock: if len(self._queue) == self._queue.maxlen: self.dropped += 1 # the deque drops the oldest @@ -92,6 +105,26 @@ def add( except Exception: self.dropped += 1 + def add_stream( + self, + provider: str, + conversation_id: str | None, + kwargs: dict[str, Any], + chunks: list[Any], + complete: bool, + customer: str | None = None, + ) -> None: + """Queue a streamed call once its stream is done: the chunks assembled into the answer + a non-streamed call would have had. A stream stopped early is kept and marked.""" + self.add( + provider, + conversation_id, + kwargs, + assemble(provider, chunks), + streamed={"complete": complete, "chunks": len(chunks)}, + customer=customer, + ) + def flush(self, timeout: float = 5.0) -> None: """Send everything queued now (tests, and at exit).""" deadline = time.monotonic() + timeout diff --git a/src/agentcompile/_conversation.py b/src/agentcompile/_conversation.py index 6a71b42..36d0327 100644 --- a/src/agentcompile/_conversation.py +++ b/src/agentcompile/_conversation.py @@ -9,21 +9,31 @@ _current: contextvars.ContextVar[str | None] = contextvars.ContextVar( "agentcompile_conversation", default=None ) +_customer: contextvars.ContextVar[str | None] = contextvars.ContextVar( + "agentcompile_customer", default=None +) @contextlib.contextmanager -def conversation(conversation_id: str) -> Iterator[None]: - """Every model call inside the block belongs to this conversation: +def conversation(conversation_id: str, customer: str | None = None) -> Iterator[None]: + """Every model call inside the block belongs to this conversation (and, optionally, this + customer: a stable id for the person, so repeat jobs can be counted per customer): - with agentcompile.conversation(ticket.id): + with agentcompile.conversation(ticket.id, customer=ticket.customer_id): run_agent(...) """ token = _current.set(conversation_id) + customer_token = _customer.set(customer) try: yield finally: + _customer.reset(customer_token) _current.reset(token) def current() -> str | None: return _current.get() + + +def current_customer() -> str | None: + return _customer.get() diff --git a/src/agentcompile/_core.py b/src/agentcompile/_core.py index 3f062d4..f1bef0c 100644 --- a/src/agentcompile/_core.py +++ b/src/agentcompile/_core.py @@ -2,10 +2,12 @@ from __future__ import annotations +import contextvars import time from typing import Any, Callable -from ._conversation import current +from ._assemble import AsyncCapturingStream, CapturingStream +from ._conversation import current, current_customer from ._decide import Decision from ._payload import payload from ._trail import Trail @@ -22,15 +24,20 @@ def __init__( if mode not in (LIVE, SHADOW): raise ValueError("mode must be 'live' or 'shadow'") self.capturer = capturer + self._customer: contextvars.ContextVar[str | None] = contextvars.ContextVar( + "agentcompile_call_customer", default=None + ) self.provider = provider self.decider = decider self.trail = trail self.mode = mode def prepare(self, kwargs: dict[str, Any]) -> tuple[str | None, dict[str, Any]]: - """(conversation id, kwargs for the real call): our keyword never reaches the model.""" + """(conversation id, kwargs for the real call): our keywords never reach the model. + The customer id rides along for capture only.""" kwargs = dict(kwargs) conversation_id = kwargs.pop("conversation_id", None) or current() + self._customer.set(kwargs.pop("customer_id", None) or current_customer()) return conversation_id, kwargs def plan( @@ -48,10 +55,27 @@ def plan( return "compiled", decision def capture( - self, conversation_id: str | None, kwargs: dict[str, Any], result: Any, stream: bool - ) -> None: - if self.capturer is not None: - self.capturer.add(self.provider, conversation_id, kwargs, result, stream) + self, + conversation_id: str | None, + kwargs: dict[str, Any], + result: Any, + stream: bool, + is_async: bool = False, + ) -> Any: + """Queue the call for capture; returns the result to hand back (a stream comes back + wrapped, so its answer is captured once the agent has read it).""" + if self.capturer is None: + return result + customer = self._customer.get() + if not stream: + self.capturer.add(self.provider, conversation_id, kwargs, result, customer=customer) + return result + capturer, provider = self.capturer, self.provider + + def done(chunks: list[Any], complete: bool) -> None: + capturer.add_stream(provider, conversation_id, kwargs, chunks, complete, customer) + + return (AsyncCapturingStream if is_async else CapturingStream)(result, done) def record( self, @@ -112,10 +136,9 @@ def create(*args: Any, **kwargs: Any) -> Any: started=started, stream=stream, ) - router.capture(conversation_id, real_kwargs, result, stream) - return result + return router.capture(conversation_id, real_kwargs, result, stream) result = original(*args, **real_kwargs) - router.capture(conversation_id, real_kwargs, result, stream) + result = router.capture(conversation_id, real_kwargs, result, stream) router.record( route=route, conversation_id=conversation_id, @@ -160,10 +183,9 @@ async def create(*args: Any, **kwargs: Any) -> Any: started=started, stream=stream, ) - router.capture(conversation_id, real_kwargs, result, stream) - return result + return router.capture(conversation_id, real_kwargs, result, stream, True) result = await original(*args, **real_kwargs) - router.capture(conversation_id, real_kwargs, result, stream) + result = router.capture(conversation_id, real_kwargs, result, stream, True) router.record( route=route, conversation_id=conversation_id, diff --git a/src/agentcompile/_outcome.py b/src/agentcompile/_outcome.py new file mode 100644 index 0000000..39415b9 --- /dev/null +++ b/src/agentcompile/_outcome.py @@ -0,0 +1,68 @@ +"""How a conversation ended, reported by you: `agentcompile.outcome(ticket.id, "resolved")`. + +AgentCompile learns only from conversations that went well; an outcome you report is the +strongest evidence it has. Sent in the background (never blocks, never raises), using the +key and address of the client you wrapped last. +""" + +from __future__ import annotations + +import threading +from datetime import datetime, timezone +from typing import Any + +import httpx + +from ._decide import CONVERSATION_HEADER, Settings, _request + +OUTCOMES = ("resolved", "escalated", "unresolved", "abandoned", "reopened", "complaint") + +_settings: list[Settings] = [] # the most recently wrapped client's, last +_threads: list[threading.Thread] = [] +sent = 0 +failed = 0 + + +def remember(settings: Settings) -> None: + _settings.append(settings) + + +def outcome(conversation_id: str, label: str, note: str | None = None) -> None: + """Report how a conversation ended: one of resolved, escalated, unresolved, abandoned, + reopened, complaint. Raises ValueError for anything else (a typo would be silently lost).""" + if label not in OUTCOMES: + raise ValueError(f"outcome must be one of {', '.join(OUTCOMES)}") + if not _settings: + return # nothing wrapped yet: nowhere to send it + record: dict[str, Any] = { + "conversation_id": conversation_id, + "outcome": label, + "timestamp": datetime.now(timezone.utc).isoformat(), + } + if note: + record["note"] = note + thread = threading.Thread(target=_send, args=(_settings[-1], record), daemon=True) + _threads.append(thread) + thread.start() + + +def _send(settings: Settings, record: dict[str, Any]) -> None: + global sent, failed + url, headers, _ = _request(settings, "openai", "outcome", {}) + headers.pop(CONVERSATION_HEADER, None) + url = url.rsplit("/v1/decide", 1)[0] + "/v1/outcome" + try: + with httpx.Client(timeout=settings.timeout) as http: + response = http.post(url, headers=headers, json={"outcomes": [record]}) + if response.status_code == 200: + sent += 1 + else: + failed += 1 + except Exception: + failed += 1 + + +def flush(timeout: float = 5.0) -> None: + for thread in list(_threads): + thread.join(timeout) + _threads.clear() diff --git a/tests/test_capture_complete.py b/tests/test_capture_complete.py new file mode 100644 index 0000000..7cbed40 --- /dev/null +++ b/tests/test_capture_complete.py @@ -0,0 +1,342 @@ +"""Capture, completed: streamed answers captured whole, the customer id, and outcomes.""" + +from __future__ import annotations + +import asyncio +import json +from typing import Any + +import anthropic +import httpx +import openai +import pytest + +import agentcompile +from agentcompile import _capture, _outcome +from agentcompile._assemble import assemble + +MESSAGES = [{"role": "user", "content": "cancel order #W1"}] + + +def _sse(events: list[tuple[str | None, Any]]) -> bytes: + out = [] + for name, data in events: + line = f"event: {name}\n" if name else "" + out.append(f"{line}data: {data if isinstance(data, str) else json.dumps(data)}\n\n") + return "".join(out).encode() + + +def _chunk(delta: dict[str, Any], finish: str | None = None) -> dict[str, Any]: + return { + "id": "s1", + "object": "chat.completion.chunk", + "created": 1, + "model": "m", + "choices": [{"index": 0, "delta": delta, "finish_reason": finish}], + } + + +OPENAI_STREAM = _sse( + [ + (None, _chunk({"role": "assistant", "content": "Let me "})), + (None, _chunk({"content": "check."})), + ( + None, + _chunk( + { + "tool_calls": [ + { + "index": 0, + "id": "t1", + "type": "function", + "function": {"name": "get_order", "arguments": '{"id":'}, + } + ] + } + ), + ), + (None, _chunk({"tool_calls": [{"index": 0, "function": {"arguments": '"#W1"}'}}]})), + (None, _chunk({}, "tool_calls")), + (None, "[DONE]"), + ] +) + +ANTHROPIC_STREAM = _sse( + [ + ( + "message_start", + { + "type": "message_start", + "message": { + "id": "msg1", + "type": "message", + "role": "assistant", + "model": "claude", + "content": [], + "stop_reason": None, + "stop_sequence": None, + "usage": {"input_tokens": 5, "output_tokens": 0}, + }, + }, + ), + ( + "content_block_start", + { + "type": "content_block_start", + "index": 0, + "content_block": {"type": "text", "text": ""}, + }, + ), + ( + "content_block_delta", + { + "type": "content_block_delta", + "index": 0, + "delta": {"type": "text_delta", "text": "On it."}, + }, + ), + ("content_block_stop", {"type": "content_block_stop", "index": 0}), + ( + "content_block_start", + { + "type": "content_block_start", + "index": 1, + "content_block": { + "type": "tool_use", + "id": "tu1", + "name": "get_order", + "input": {}, + }, + }, + ), + ( + "content_block_delta", + { + "type": "content_block_delta", + "index": 1, + "delta": {"type": "input_json_delta", "partial_json": '{"id": "#W1"}'}, + }, + ), + ("content_block_stop", {"type": "content_block_stop", "index": 1}), + ( + "message_delta", + { + "type": "message_delta", + "delta": {"stop_reason": "tool_use", "stop_sequence": None}, + "usage": {"output_tokens": 9}, + }, + ), + ("message_stop", {"type": "message_stop"}), + ] +) + + +class Service: + def __init__(self) -> None: + self.captured: list[dict[str, Any]] = [] + self.outcomes: list[dict[str, Any]] = [] + + def __call__(self, request: httpx.Request) -> httpx.Response: + body = json.loads(request.content or b"{}") + if request.url.path == "/v1/capture": + self.captured += body["exchanges"] + elif request.url.path == "/v1/outcome": + self.outcomes += body["outcomes"] + else: + return httpx.Response(200, json={"action": "forward", "reason": "x"}) + return httpx.Response(200, json={}) + + +def _anthropic_provider(body: bytes) -> Any: + """anthropic 1.x ships its own httpx fork (httpx2), as in test_wrap.py.""" + try: + import httpx2 as hx + except ImportError: # older anthropic: plain httpx + hx = httpx + headers = {"content-type": "text/event-stream"} + return hx.Client( + transport=hx.MockTransport(lambda r: hx.Response(200, content=body, headers=headers)) + ) + + +def _provider(body: bytes) -> httpx.MockTransport: + return httpx.MockTransport( + lambda r: httpx.Response( + 200, content=body, headers={"content-type": "text/event-stream"} + ) + ) + + +@pytest.fixture(autouse=True) +def _env(monkeypatch: pytest.MonkeyPatch) -> Any: + monkeypatch.setenv("AGENTCOMPILE_SCRUB_KEY", "k") + yield + _capture._ALL.clear() + + +def _openai(service: Service, body: bytes = OPENAI_STREAM) -> Any: + real = openai.OpenAI( + api_key="sk", + base_url="http://p.test/v1", + http_client=httpx.Client(transport=_provider(body)), + ) + return agentcompile.wrap( + real, + base_url="http://ac.test", + trail=False, + capture=True, + http_client=httpx.Client(transport=httpx.MockTransport(service)), + ) + + +def test_a_streamed_openai_answer_is_captured_whole() -> None: + service = Service() + stream = _openai(service).chat.completions.create( + model="m", messages=MESSAGES, stream=True, conversation_id="c1" + ) + seen = [chunk.choices[0].delta for chunk in stream] # the agent reads it as always + assert len(seen) == 5 + agentcompile.flush(5) + [record] = service.captured + message = record["response"]["choices"][0]["message"] + assert message["content"] == "Let me check." + assert message["tool_calls"][0]["function"] == { + "name": "get_order", + "arguments": '{"id":"#W1"}', + } + assert record["response"]["choices"][0]["finish_reason"] == "tool_calls" + assert record["stream"] is True and record["stream_complete"] is True + + +def test_a_stream_the_agent_stops_reading_is_marked_incomplete() -> None: + service = Service() + stream = _openai(service).chat.completions.create( + model="m", messages=MESSAGES, stream=True, conversation_id="c1" + ) + for _ in stream: + break + stream.close() + agentcompile.flush(5) + [record] = service.captured + assert record["stream_complete"] is False + + +def test_a_streamed_anthropic_answer_is_captured_whole() -> None: + service = Service() + real = anthropic.Anthropic( + api_key="sk", + base_url="http://p.test", + http_client=_anthropic_provider(ANTHROPIC_STREAM), + ) + client = agentcompile.wrap( + real, + base_url="http://ac.test", + trail=False, + capture=True, + http_client=httpx.Client(transport=httpx.MockTransport(service)), + ) + with client.messages.create( + model="claude", max_tokens=50, messages=MESSAGES, stream=True, conversation_id="c1" + ) as stream: + kinds = [event.type for event in stream] + assert kinds[0] == "message_start" and kinds[-1] == "message_stop" + agentcompile.flush(5) + [record] = service.captured + response = record["response"] + assert response["content"][0] == {"type": "text", "text": "On it."} + assert response["content"][1]["input"] == {"id": "#W1"} + assert response["stop_reason"] == "tool_use" and response["usage"]["output_tokens"] == 9 + + +def test_async_streams_are_captured_too() -> None: + service = Service() + + async def run() -> list[Any]: + real = openai.AsyncOpenAI( + api_key="sk", + base_url="http://p.test/v1", + http_client=httpx.AsyncClient(transport=_provider(OPENAI_STREAM)), + ) + client = agentcompile.wrap(real, base_url="http://ac.test", trail=False, capture=True) + _capture._ALL[-1]._http = httpx.Client(transport=httpx.MockTransport(service)) + stream = await client.chat.completions.create( + model="m", messages=MESSAGES, stream=True, conversation_id="c1" + ) + return [chunk async for chunk in stream] + + chunks = asyncio.run(run()) + assert len(chunks) == 5 + agentcompile.flush(5) + [record] = service.captured + assert record["response"]["choices"][0]["message"]["content"] == "Let me check." + + +def test_a_compiled_stream_is_captured_whole() -> None: + class Compiling(Service): + def __call__(self, request: httpx.Request) -> httpx.Response: + if request.url.path == "/v1/decide": + return httpx.Response(200, json={"action": "say", "text": "Done."}) + return super().__call__(request) + + service = Compiling() + stream = _openai(service).chat.completions.create( + model="m", messages=MESSAGES, stream=True, conversation_id="c1" + ) + list(stream) + agentcompile.flush(5) + [record] = service.captured + assert record["response"]["choices"][0]["message"]["content"] == "Done." + + +def test_the_customer_id_is_captured_scrubbed() -> None: + service = Service() + client = _openai(service, OPENAI_STREAM) + with agentcompile.conversation("c1", customer="mia@example.com"): + list(client.chat.completions.create(model="m", messages=MESSAGES, stream=True)) + list( + client.chat.completions.create( + model="m", + messages=MESSAGES, + stream=True, + conversation_id="c2", + customer_id="cust_42", + ) + ) + agentcompile.flush(5) + first, second = service.captured + assert first["end_user"].startswith(" None: + service = Service() + _openai(service) + _outcome._settings[-1] = agentcompile.Settings("http://ac.test", "ack_acme.k", None, 2.0) + + real_client = httpx.Client + + class Routed(real_client): # type: ignore[misc,valid-type] + def __init__(self, **kw: Any) -> None: + super().__init__(transport=httpx.MockTransport(service), **kw) + + import agentcompile._outcome as module + + module.httpx.Client = Routed # type: ignore[misc] + try: + agentcompile.outcome("c1", "escalated", note="customer asked for a human") + agentcompile.flush(5) + finally: + module.httpx.Client = real_client # type: ignore[misc] + [record] = service.outcomes + assert record["conversation_id"] == "c1" and record["outcome"] == "escalated" + assert record["note"] == "customer asked for a human" + + +def test_an_unknown_outcome_is_refused() -> None: + with pytest.raises(ValueError): + agentcompile.outcome("c1", "fine") + + +def test_assembly_never_breaks_the_agent() -> None: + assert assemble("openai", [{"choices": "nonsense"}]) is None + assert assemble("anthropic", [{"type": "content_block_delta"}]) is None