Setting the file. One moment. def parse_user_turn
— line 68
This file
- Number
- 7.9
- Position
- 9 of 18
- Type
- Python
- Size
- 7 KB
- Lines
- 211
references/multi-turn-simulation/runner.py
Python·211 lines·7 KB
12
"""The simulator failed; do not score this as Harness failure."""
13
14 def __init__(self, message: str, *, evidence: Optional[object] = None) -> None:
15 super().__init__(message)
16 self.evidence = evidence
17
18
19@dataclass(frozen=True)
20class Turn:
21 role: Literal["user", "assistant"]
22 content: str
23 origin: Literal["instruction", "scripted", "harness", "simulator"]
24 decision_id: Optional[str] = None
25 timestamp: str = ""
26
27
28@dataclass(frozen=True)
29class UserTurn:
30 message: Optional[str]
31 stop: bool
32
33
34@dataclass(frozen=True)
35class ConversationResult:
36 turns: tuple[Turn, ...]
37 termination: Literal[
38 "script_finished", "user_stop", "turn_limit", "simulator_error"
39 ]
40 harness_calls: int
41 simulator_calls: int
42 error: Optional[str] = None
43
44
45class ConversationRunError(RuntimeError):
46 """A simulator failure with delivered turns retained for audit."""
47
48 def __init__(
49 self,
50 message: str,
51 *,
52 result: ConversationResult,
53 simulator_evidence: Optional[object] = None,
54 ) -> None:
55 super().__init__(message)
56 self.result = result
57 self.simulator_evidence = simulator_evidence
58
59
60class AgentSession(Protocol):
61 async def send(self, user_message: str) -> str: ...
62
63
64class SimulatedUser(Protocol):
65 async def reply(self, transcript: tuple[Turn, ...]) -> UserTurn: ...
66
67
68def parse_user_turn(raw: str) -> UserTurn:
69 """Parse either a delivered reply or a message-free stop decision."""
70 try:
71 value = json.loads(raw)
72 except (TypeError, json.JSONDecodeError) as error:
73 raise SimulatorProtocolError("simulator output is not valid JSON") from error
74 if not isinstance(value, dict) or type(value.get("stop")) is not bool:
75 raise SimulatorProtocolError("simulator output must contain a boolean stop")
76 if value["stop"] is True:
77 if set(value) != {"stop"}:
78 raise SimulatorProtocolError("a stop decision must not contain a message")
79 return UserTurn(message=None, stop=True)
80 if set(value) != {"message", "stop"}:
81 raise SimulatorProtocolError("a reply must contain only message and stop")
82 message = value["message"]
83 if not isinstance(message, str) or not 1 <= len(message.strip()) <= 2_000:
84 raise SimulatorProtocolError("simulator message is invalid")
85 return UserTurn(message=message, stop=False)
86
87
88def validate_user_turn(value: object) -> UserTurn:
89 """Validate turns from custom simulators as strictly as model output."""
90 if not isinstance(value, UserTurn) or type(value.stop) is not bool:
91 raise SimulatorProtocolError("simulator did not return a valid UserTurn")
92 if value.stop:
93 if value.message is not None:
94 raise SimulatorProtocolError("a stop decision must not contain a message")
95 elif not isinstance(value.message, str) or not 1 <= len(value.message.strip()) <= 2_000:
96 raise SimulatorProtocolError("simulator message is invalid")
97 return value
98
99
100def _timestamp() -> str:
101 return datetime.now(timezone.utc).isoformat().replace("+00:00", "Z")
102
103
104def _message(value: object, source: str) -> str:
105 if not isinstance(value, str) or not 1 <= len(value.strip()) <= 100_000:
106 raise ValueError(f"{source} message is invalid")
107 return value
108
109
110async def run_scripted_conversation(
111 *,
112 first_message: str,
113 followups: tuple[str, ...],
114 session: AgentSession,
115) -> ConversationResult:
116 """Send known user turns sequentially through one Harness session."""
117 turns: list[Turn] = []
118 for index, raw_message in enumerate((first_message, *followups)):
119 message = _message(raw_message, "user")
120 turns.append(
121 Turn("user", message, "instruction" if index == 0 else "scripted", timestamp=_timestamp())
122 )
123 reply = _message(await session.send(message), "assistant")
124 turns.append(Turn("assistant", reply, "harness", timestamp=_timestamp()))
125 return ConversationResult(
126 tuple(turns), "script_finished", len(followups) + 1, 0
127 )
128
129
130async def run_llm_user_conversation(
131 *,
132 first_message: str,
133 session: AgentSession,
134 simulated_user: SimulatedUser,
135 max_turns: int = 8,
136) -> ConversationResult:
137 """Alternate one Harness session with an external LLM user."""
138 if max_turns < 1:
139 raise ValueError("max_turns must be positive")
140
141 user_message = _message(first_message, "first user")
142 turns = [Turn("user", user_message, "instruction", timestamp=_timestamp())]
143 simulator_calls = 0
144
145 for harness_calls in range(1, max_turns + 1):
146 assistant_message = _message(await session.send(user_message), "assistant")
147 turns.append(Turn("assistant", assistant_message, "harness", timestamp=_timestamp()))
148
149 simulator_calls += 1
150 try:
151 reply = await simulated_user.reply(tuple(turns))
152 except SimulatorProtocolError as error:
153 result = ConversationResult(
154 tuple(turns),
155 "simulator_error",
156 harness_calls,
157 simulator_calls,
158 str(error),
159 )
160 raise ConversationRunError(
161 str(error), result=result, simulator_evidence=error.evidence
162 ) from error
163 except Exception as error:
164 message = f"simulator failed: {type(error).__name__}"
165 result = ConversationResult(
166 tuple(turns),
167 "simulator_error",
168 harness_calls,
169 simulator_calls,
170 message,
171 )
172 raise ConversationRunError(
173 message,
174 result=result,
175 simulator_evidence={"error": type(error).__name__},
176 ) from error
177 try:
178 reply = validate_user_turn(reply)
179 except SimulatorProtocolError as error:
180 message = str(error)
181 result = ConversationResult(
182 tuple(turns),
183 "simulator_error",
184 harness_calls,
185 simulator_calls,
186 message,
187 )
188 raise ConversationRunError(message, result=result) from error
189
190 if reply.stop:
191 return ConversationResult(
192 tuple(turns), "user_stop", harness_calls, simulator_calls
193 )
194
195 user_message = reply.message or ""
196 turns.append(
197 Turn(
198 "user",
199 user_message,
200 "simulator",
201 f"sim-{simulator_calls:03d}",
202 _timestamp(),
203 )
204 )
205
206 if harness_calls == max_turns:
207 return ConversationResult(
208 tuple(turns), "turn_limit", harness_calls, simulator_calls
209 )
210
211 raise AssertionError("conversation loop terminated unexpectedly")