def session_pre_process(context):
resp = context.http.post(
context.vars["base_url"] + "/sessions",
headers={"Authorization": f"Bearer {context.auth['token']}"},
)
return SessionPreProcessResult(session_state={"session_id": resp.json()["id"]})
def pre_process(context, inference_input):
return PreProcessResult(
url=context.vars["base_url"] + "/chat",
headers={"Authorization": f"Bearer {context.auth['token']}"},
json_body={
"session_id": context.session["session_id"],
"message": inference_input.prompt,
},
)
Stateless Target
The target does not track conversation history. Resend the full conversation history
on each turn using inference_input.previous_messages.
def pre_process(context, inference_input):
messages = [
{"role": m.role, "content": m.content}
for m in inference_input.previous_messages
]
messages.append({"role": "user", "content": inference_input.prompt})
return PreProcessResult(
url=context.vars["endpoint"],
headers={"Authorization": f"Bearer {context.secrets['api_key']}"},
json_body={"messages": messages},
)
def post_process(context, raw_response):
return PostProcessResult(output=raw_response.json_body["reply"])
Updating Session State Mid-Conversation
To update session state during a conversation, return the updated state in the
session_state field of
PostProcessResult or
CallTargetResult. The returned value
replaces
context.session for the next turn. This does not affect
inference_input.previous_messages, which AI Red Teaming builds
and manages.
