Coverage for src / quber / playground / answers / expectation.py: 50%

1"""Answering to a stated expectation rather than an inferred one. 

2 

3The union in `shapes` lets the model decide whether a question wants a figure 

4or an explanation, and it decides well — but "well" is not "predictably", and a 

5caller that wanted prose and got a figure has no recourse except to rephrase. 

6So the expectation is something the caller states: 

7 

8 auto the model chooses the payload from the union 

9 value a figure, whatever the question looked like 

10 text an explanation, whatever the question looked like 

11 

12Every path returns the same `Answer`, so stating an expectation changes which 

13shape comes back and nothing else about the contract. `value` and `text` go 

14through `declared` — the caller-supplied-schema path — which is exactly what 

15stating an expectation means. 

16 

17A forced shape can still fail honestly: asking for a figure where the context 

18holds none returns `unanswerable` rather than inventing one. 

19""" 

20 

21from __future__ import annotations 

22 

23from typing import AsyncIterator, List, Literal, Optional, Tuple 

24 

25from quber.playground.answers import declared, fixed, streaming 

26from quber.playground.answers.document import DocumentIdentity 

27from quber.playground.answers.shapes import Answer, Prose, Scalar, Unanswerable 

28from quber.playground.retrieval import RetrievedChunk 

29 

30Want = Literal["auto", "value", "text"] 

31 

32 

33def scalar_answer(result: declared.DeclaredScalar) -> Answer: 

34 if result.not_found: 

35 return Answer(payload=Unanswerable(reason=result.not_found), cited_ids=result.cited_ids) 

36 return Answer( 

37 payload=Scalar(value=result.value, unit=result.unit, period=result.period), 

38 cited_ids=result.cited_ids, 

39 ) 

40 

41 

42def prose_answer(result: declared.DeclaredProse) -> Answer: 

43 if result.not_found and not result.text: 

44 return Answer(payload=Unanswerable(reason=result.not_found), cited_ids=result.cited_ids) 

45 return Answer(payload=Prose(text=result.text), cited_ids=result.cited_ids) 

46 

47 

48async def answer( 

49 question: str, 

50 chunks: List[RetrievedChunk], 

51 want: Want = "auto", 

52 document: Optional[DocumentIdentity] = None, 

53) -> Answer: 

54 if want == "value": 

55 return scalar_answer(await declared.answer(question, chunks, declared.DeclaredScalar, document)) 

56 if want == "text": 

57 return prose_answer(await declared.answer(question, chunks, declared.DeclaredProse, document)) 

58 return await fixed.answer(question, chunks, document) 

59 

60 

61async def answer_stream( 

62 question: str, 

63 chunks: List[RetrievedChunk], 

64 want: Want = "auto", 

65 document: Optional[DocumentIdentity] = None, 

66) -> AsyncIterator[Tuple[str, object]]: 

67 """Yield ("delta", text) while readable text grows, then ("final", Answer).""" 

68 if want == "text": 

69 async for kind, payload in declared.answer_stream( 

70 question, chunks, declared.DeclaredProse, "text", document 

71 ): 

72 if kind == "delta": 

73 yield kind, payload 

74 else: 

75 assert isinstance(payload, declared.DeclaredProse) 

76 yield "final", prose_answer(payload) 

77 return 

78 if want == "value": 

79 # Nothing to animate in a figure; one final event keeps the caller's 

80 # event contract identical across the three expectations. 

81 yield "final", await answer(question, chunks, "value", document) 

82 return 

83 async for kind, payload in streaming.answer_stream(question, chunks, document): 

84 yield kind, payload