interlens.participant.participants.api_client¶

def submit_batch(self, requests: list[dict], *, poll_interval: float = 30.0) -> list[str]:
	"""Anthropic **Message Batches API**: one ``messages.batches.create`` submits every request (tagged with a
	positional ``custom_id``), then poll ``retrieve`` until ``processing_status == 'ended'`` and stream
	``results`` back, reassembled into input order. A non-succeeded per-request result raises."""
	batch = self._client.messages.batches.create(requests=[
		{"custom_id": f"req-{i}",
		 "params": {"model": r["model"], "max_tokens": r["max_tokens"], "temperature": r["temperature"],
		            "messages": r["messages"],
		            **({"system": r["system"]} if r.get("system") else {})}}
		for i, r in enumerate(requests)])
	while self._client.messages.batches.retrieve(batch.id).processing_status != "ended":
		time.sleep(poll_interval)  # await external batch completion (not a fixed delay)
	texts: dict[str, str] = {}
	for entry in self._client.messages.batches.results(batch.id):
		if entry.result.type != "succeeded":
			raise RuntimeError(f"Anthropic batch request {entry.custom_id} did not succeed: {entry.result.type}")
		msg = entry.result.message
		texts[entry.custom_id] = "".join(b.text for b in msg.content if getattr(b, "type", None) == "text")
	return [texts[f"req-{i}"] for i in range(len(requests))]

def submit_batch(self, requests: list[dict], *, poll_interval: float = 30.0) -> list[str]:
	"""OpenAI **Batch API**: upload a JSONL of ``/v1/chat/completions`` requests (positional ``custom_id``),
	``batches.create`` with a 24h window, poll until ``status == 'completed'``, then download + parse the
	output file back into input order. A failed/expired/cancelled batch raises."""
	import io
	import json

	lines = [json.dumps({
		"custom_id": f"req-{i}", "method": "POST", "url": "/v1/chat/completions",
		"body": {"model": r["model"], self._tokens_param: r["max_tokens"], "temperature": r["temperature"],
		         "messages": self._full_messages(r.get("system"), r["messages"])}})
		for i, r in enumerate(requests)]
	upload = self._client.files.create(
		file=("batch.jsonl", io.BytesIO("\n".join(lines).encode())), purpose="batch")
	batch = self._client.batches.create(
		input_file_id=upload.id, endpoint="/v1/chat/completions", completion_window="24h")
	while True:
		batch = self._client.batches.retrieve(batch.id)
		if batch.status == "completed":
			break
		if batch.status in ("failed", "expired", "cancelled", "cancelling"):
			raise RuntimeError(f"OpenAI batch {batch.id} ended as {batch.status}")
		time.sleep(poll_interval)  # await external batch completion (not a fixed delay)
	texts: dict[str, str] = {}
	content = self._client.files.content(batch.output_file_id).text
	for line in content.splitlines():
		if not line.strip():
			continue
		obj = json.loads(line)
		texts[obj["custom_id"]] = obj["response"]["body"]["choices"][0]["message"]["content"] or ""
	return [texts[f"req-{i}"] for i in range(len(requests))]