tiennm99/DocsGPT
1
0 Fork 0
mirror of https://github.com/tiennm99/DocsGPT.git synced 2026-10-12 14:14:50 +00:00
Code Issues Packages Projects Releases Wiki Activity

Merge pull request #2523 from arc53/feat/v1-response-format

Browse Source 
feat(v1): response_format / response_schema (Structured Outputs) on /v1/chat/completions
This commit is contained in:
Alex authored and GitHub committed 2026-06-04 14:57:53 +01:00
2 parents c05b457b55 45cae8f6b8
commit 64e7210cf1
9 files changed
+961 -70
Download Patch File Download Diff File Expand all files Collapse all files

No files matched your search

application/agents/base.py
+37 -2
View File
@@ -33,6 +33,10 @@ class BaseAgent(ABC):
decoded_token: Optional[Dict] = None,
attachments: Optional[List[Dict]] = None,
json_schema: Optional[Dict] = None,
json_schema_strict: bool = True,
json_object: bool = False,
llm_params: Optional[Dict] = None,
multimodal_content: Optional[List] = None,
limited_token_mode: Optional[bool] = False,
token_limit: Optional[int] = settings.DEFAULT_AGENT_LIMITS["token_limit"],
limited_request_mode: Optional[bool] = False,
@@ -108,6 +112,17 @@ class BaseAgent(ABC):
self.json_schema = normalize_json_schema_payload(json_schema)
except JsonSchemaValidationError as exc:
logger.warning("Ignoring invalid JSON schema payload: %s", exc)
# Per-request structured-output controls (OpenAI-compatible):
# ``json_schema_strict`` mirrors response_format.json_schema.strict;
# ``json_object`` mirrors response_format {"type":"json_object"}.
self.json_schema_strict = json_schema_strict
self.json_object = json_object
# OpenAI sampling params forwarded from the request (temperature,
# max_tokens, top_p, ...). Empty when the caller sent none.
self.llm_params = llm_params or {}
# Full OpenAI content array (text + image_url parts) for the current
# user turn, when the request was multimodal; None otherwise.
self.multimodal_content = multimodal_content
self.limited_token_mode = limited_token_mode
self.token_limit = token_limit
self.limited_request_mode = limited_request_mode
@@ -501,7 +516,15 @@ class BaseAgent(ABC):
"tool_call_id": call_id,
"content": result_str,
})
messages.append({"role": "user", "content": query})
# When the request was multimodal, send the full content array (text +
# image_url parts) so images reach the model; the text-only `query` above
# is used only for token budgeting / retrieval.
user_content = (
self.multimodal_content
if getattr(self, "multimodal_content", None)
else query
)
messages.append({"role": "user", "content": user_content})
return messages
def _truncate_history_to_fit(
@@ -572,13 +595,21 @@ class BaseAgent(ABC):
and self.llm._supports_structured_output()
):
structured_format = self.llm.prepare_structured_output_format(
self.json_schema
self.json_schema, strict=getattr(self, "json_schema_strict", True)
)
if structured_format:
if self.llm_name == "openai":
gen_kwargs["response_format"] = structured_format
elif self.llm_name == "google":
gen_kwargs["response_schema"] = structured_format
elif (
getattr(self, "json_object", False)
and self.llm_name == "openai"
and hasattr(self.llm, "_supports_structured_output")
and self.llm._supports_structured_output()
):
# OpenAI json_object mode: guarantee valid JSON, no schema enforcement.
gen_kwargs["response_format"] = {"type": "json_object"}
if (
settings.OPENAI_RESPONSES_STORE
and hasattr(self.llm, "_uses_responses_api")
@@ -587,6 +618,10 @@ class BaseAgent(ABC):
previous_response_id = self._previous_response_id()
if previous_response_id:
gen_kwargs["previous_response_id"] = previous_response_id
# Forward OpenAI sampling params (temperature, max_tokens, top_p, ...).
if self.llm_params:
gen_kwargs.update(self.llm_params)
resp = self.llm.gen_stream(**gen_kwargs)
if log_context:
application/api/answer/services/stream_processor.py
+88
View File
@@ -172,6 +172,63 @@ class StreamProcessor:
tools_data=tools_data,
)
def build_continuation_from_messages(self, messages, tool_actions):
"""Rebuild a tool continuation from the request messages (STATELESS).
OpenAI-compatible clients (opencode, etc.) resend the full conversation
-- system, user, assistant(tool_calls), tool(results) -- but carry no
conversation_id, so there is no server-side ``pending_tool_state`` to
load. Reconstruct the agent + continuation context directly from the
resent messages and return the same tuple as ``resume_from_tool_actions``:
(agent, messages, tools_dict, pending_tool_calls, tool_actions,
reasoning_content).
"""
# Locate the last assistant message that issued tool calls.
pending_idx = None
for i in range(len(messages) - 1, -1, -1):
m = messages[i]
if m.get("role") == "assistant" and m.get("tool_calls"):
pending_idx = i
break
if pending_idx is None:
raise ValueError(
"No assistant message with tool_calls found for continuation"
)
pending_tool_calls = []
for tc in messages[pending_idx].get("tool_calls") or []:
fn = tc.get("function") or {}
raw_args = fn.get("arguments")
try:
args = (
json.loads(raw_args)
if isinstance(raw_args, str)
else (raw_args or {})
)
except (json.JSONDecodeError, TypeError):
args = {}
name = fn.get("name", "")
pending_tool_calls.append(
{
"call_id": tc.get("id", ""),
"name": name,
"tool_name": name,
"action_name": name,
"llm_name": name,
"arguments": args,
}
)
# The conversation up to (but not including) the assistant tool_calls;
# gen_continuation re-appends the assistant message + tool results.
prior_messages = [dict(m) for m in messages[:pending_idx]]
# Build a normal agent (config / LLM / client tools), no new question.
agent = self.build_agent("")
tools_dict = agent.tool_executor.get_tools()
return agent, prior_messages, tools_dict, pending_tool_calls, tool_actions, ""
def _load_conversation_history(self):
"""Load conversation history either from DB or request"""
if self.conversation_id and self.initial_user_id:
@@ -647,6 +704,21 @@ class StreamProcessor:
}
)
# Per-request structured output: a ``response_format`` / ``response_schema``
# in the request (surfaced by the v1 translator as ``json_schema``) overrides
# the agent's configured schema for this call. Invalid schemas are ignored
# downstream by the agent (normalize_json_schema_payload).
request_json_schema = self.data.get("json_schema")
if request_json_schema is not None:
self.agent_config["json_schema"] = request_json_schema
if self.data.get("json_schema_strict") is not None:
self.agent_config["json_schema_strict"] = self.data.get("json_schema_strict")
if self.data.get("json_object"):
self.agent_config["json_object"] = True
# An explicit json_object request beats an agent-configured schema
# (otherwise the configured json_schema would silently override it).
self.agent_config["json_schema"] = None
def _configure_retriever(self):
"""Assemble retriever config; agent's values are authoritative when bound."""
# BYOM scope: owner for shared-agent BYOM, caller for own BYOM,
@@ -1240,6 +1312,18 @@ class StreamProcessor:
if client_tools:
tool_executor.client_tools = client_tools
# OpenAI-style image_url content parts are only understood by the
# OpenAI-family providers; drop multimodal content for others (Google,
# Anthropic, ...) so a multimodal request degrades to text rather than
# erroring upstream.
from application.llm.openai import OpenAILLM
request_multimodal = (
self.data.get("multimodal_content")
if isinstance(llm, OpenAILLM)
else None
)
agent_kwargs = {
"endpoint": "stream",
"llm_name": provider or settings.LLM_PROVIDER,
@@ -1254,6 +1338,10 @@ class StreamProcessor:
"decoded_token": self.decoded_token,
"attachments": self.attachments,
"json_schema": self.agent_config.get("json_schema"),
"json_schema_strict": self.agent_config.get("json_schema_strict", True),
"json_object": self.agent_config.get("json_object", False),
"llm_params": self.data.get("llm_params") or {},
"multimodal_content": request_multimodal,
"compressed_summary": self.compressed_summary,
"llm": llm,
"llm_handler": llm_handler,
application/api/v1/routes.py
+40 -16
View File
@@ -106,21 +106,32 @@ def chat_completions():
if internal_data.get("tool_actions"):
# Continuation mode
conversation_id = internal_data.get("conversation_id")
if not conversation_id:
return make_response(
jsonify({"error": {"message": "conversation_id required for tool continuation", "type": "invalid_request"}}),
400,
if conversation_id:
(
agent,
messages,
tools_dict,
pending_tool_calls,
tool_actions,
reasoning_content,
) = processor.resume_from_tool_actions(
internal_data["tool_actions"], conversation_id
)
else:
# Stateless continuation: OpenAI-compatible clients (opencode,
# etc.) resend the full messages array but no conversation_id,
# so rebuild the agent + pending calls from the request itself.
(
agent,
messages,
tools_dict,
pending_tool_calls,
tool_actions,
reasoning_content,
) = processor.build_continuation_from_messages(
internal_data.get("messages", []),
internal_data["tool_actions"],
)
(
agent,
messages,
tools_dict,
pending_tool_calls,
tool_actions,
reasoning_content,
) = processor.resume_from_tool_actions(
internal_data["tool_actions"], conversation_id
)
continuation = {
"messages": messages,
"tools_dict": tools_dict,
@@ -147,6 +158,12 @@ def chat_completions():
return usage_error
should_save_conversation = bool(internal_data.get("save_conversation", False))
# Only strip leaked reasoning from content for structured requests -- the
# only path where models echo reasoning into content -- so legitimate
# answers that mention the marker text are never corrupted.
strip_reasoning_leak = bool(
internal_data.get("json_schema") or internal_data.get("json_object")
)
if is_stream:
return Response(
@@ -158,6 +175,7 @@ def chat_completions():
model_name,
continuation,
should_save_conversation,
strip_reasoning_leak,
),
mimetype="text/event-stream",
headers={
@@ -174,6 +192,7 @@ def chat_completions():
model_name,
continuation,
should_save_conversation,
strip_reasoning_leak,
)
except ValueError as e:
@@ -204,6 +223,7 @@ def _stream_response(
model_name: str,
continuation: Optional[Dict],
should_save_conversation: bool,
strip_reasoning_leak: bool = False,
) -> Generator[str, None, None]:
"""Generate translated SSE chunks for streaming response."""
completion_id = f"chatcmpl-{int(time.time())}"
@@ -247,11 +267,13 @@ def _stream_response(
# Update completion_id when we get the conversation id
if event_data.get("type") == "id":
conv_id = event_data.get("id", "")
if conv_id:
if conv_id and conv_id != "None":
completion_id = f"chatcmpl-{conv_id}"
# Translate to standard format
translated = translate_stream_event(event_data, completion_id, model_name)
translated = translate_stream_event(
event_data, completion_id, model_name, strip_reasoning_leak
)
for chunk in translated:
yield chunk
@@ -264,6 +286,7 @@ def _non_stream_response(
model_name: str,
continuation: Optional[Dict],
should_save_conversation: bool,
strip_reasoning_leak: bool = False,
) -> Response:
"""Collect full response and return as single JSON."""
stream = helper.complete_stream(
@@ -298,6 +321,7 @@ def _non_stream_response(
thought=result["thought"] or "",
model_name=model_name,
pending_tool_calls=pending,
strip_reasoning_leak=strip_reasoning_leak,
)
return make_response(jsonify(response), 200)
application/api/v1/translator.py
+223 -39
View File
@@ -7,9 +7,49 @@ This module handles:
"""
import json
import re
import time
from typing import Any, Dict, List, Optional
# Some upstream models/proxies echo their reasoning into ``content`` as
# stringified ``{'type': 'thought', 'thought': '...'}`` event reprs (instead of
# using the separate reasoning channel) — most visibly when ``response_format``
# is set. OpenAI's API never puts reasoning in ``content``, so for the
# OpenAI-compatible endpoint we strip these and reroute them to
# ``reasoning_content`` to keep ``content`` clean and compatible.
# The thought value is a Python string repr: single-quoted, or double-quoted when
# the token contains an apostrophe (e.g. "'ll"). Match the full quoted value
# (honoring escapes) so tokens containing ``}`` or newlines don't truncate the
# match and leave stray ``'}`` tails in the content.
_LEAKED_THOUGHT_RE = re.compile(
r"""\{'type': 'thought', 'thought': ('(?:[^'\\]|\\.)*'|"(?:[^"\\]|\\.)*")\}""",
re.DOTALL,
)
def _strip_repr_quotes(value: str) -> str:
value = value.strip()
if len(value) >= 2 and value[0] in "\"'" and value[-1] == value[0]:
return value[1:-1]
return value
def _split_leaked_reasoning(content: Optional[str]) -> tuple:
"""Return ``(clean_content, leaked_reasoning)``.
``clean_content`` has any stringified thought-event reprs removed;
``leaked_reasoning`` is the concatenated reasoning text that was extracted.
A no-op (returns the input unchanged) when no leak markers are present.
"""
if not content or "'type': 'thought'" not in content:
return content, ""
extracted: List[str] = []
cleaned = _LEAKED_THOUGHT_RE.sub(
lambda m: (extracted.append(_strip_repr_quotes(m.group(1))) or ""), content
)
return cleaned, "".join(extracted)
def _get_client_tool_name(tc: Dict) -> str:
"""Return the original tool name for client-facing responses.
@@ -80,6 +120,28 @@ def extract_conversation_id(messages: List[Dict]) -> Optional[str]:
return None
def content_to_text(content: Any) -> str:
"""Flatten an OpenAI message ``content`` to plain text.
``content`` may be a string or a list of typed parts
(``{"type":"text",...}`` / ``{"type":"image_url",...}`` / ...). Only text
parts contribute; image/other parts are dropped here. The full content
array is preserved separately (see ``multimodal_content``) so images still
reach the model in the final user message.
"""
if isinstance(content, str):
return content
if isinstance(content, list):
out = []
for part in content:
if isinstance(part, dict) and part.get("type") == "text":
out.append(part.get("text", "") or "")
elif isinstance(part, str):
out.append(part)
return "\n".join(out)
return "" if content is None else str(content)
def extract_system_prompt(messages: List[Dict]) -> Optional[str]:
"""Extract the first system message content from the messages array.
@@ -87,7 +149,7 @@ def extract_system_prompt(messages: List[Dict]) -> Optional[str]:
"""
for msg in messages:
if msg.get("role") == "system":
return msg.get("content", "")
return content_to_text(msg.get("content", ""))
return None
@@ -107,9 +169,9 @@ def convert_history(messages: List[Dict]) -> List[Dict]:
if msg.get("role") == "user":
# Look ahead for assistant response
if i + 1 < len(messages) and messages[i + 1].get("role") == "assistant":
content = messages[i + 1].get("content") or ""
content = content_to_text(messages[i + 1].get("content") or "")
history.append({
"prompt": msg.get("content", ""),
"prompt": content_to_text(msg.get("content", "")),
"response": content,
})
i += 2
@@ -121,6 +183,37 @@ def convert_history(messages: List[Dict]) -> List[Dict]:
return history
def extract_response_schema(data: Dict[str, Any]) -> Optional[Dict[str, Any]]:
"""Extract a JSON schema for structured output from a chat-completions request.
Supports two request shapes:
- OpenAI ``response_format``:
``{"type": "json_schema", "json_schema": {"name": ..., "schema": {...}}}``
(a bare schema under ``json_schema`` is also tolerated).
- ``response_schema`` convenience field: a raw JSON Schema object, or a
``{"schema": {...}}`` wrapper.
Returns a raw JSON Schema object, or None. ``response_format``
``{"type": "json_object"}`` carries no schema to enforce and yields None
(the model is still steered by the system prompt).
"""
response_schema = data.get("response_schema")
if isinstance(response_schema, dict) and response_schema:
inner = response_schema.get("schema")
return inner if isinstance(inner, dict) else response_schema
response_format = data.get("response_format")
if isinstance(response_format, dict) and response_format.get("type") == "json_schema":
json_schema = response_format.get("json_schema")
if isinstance(json_schema, dict):
schema = json_schema.get("schema")
if isinstance(schema, dict):
return schema
if "type" in json_schema:
return json_schema
return None
def translate_request(
data: Dict[str, Any], api_key: str
) -> Dict[str, Any]:
@@ -134,6 +227,26 @@ def translate_request(
Dict suitable for passing to ``StreamProcessor``.
"""
messages = data.get("messages", [])
response_schema = extract_response_schema(data)
_rf = data.get("response_format")
_rf = _rf if isinstance(_rf, dict) else {}
# OpenAI Structured Outputs default to strict; honor an explicit strict:false.
json_schema_strict = bool((_rf.get("json_schema") or {}).get("strict", True))
json_object_mode = _rf.get("type") == "json_object"
# OpenAI sampling params, forwarded to the LLM gen call (the agent otherwise
# uses its configured defaults).
sampling_params = {}
for _k in (
"temperature", "max_tokens", "max_completion_tokens",
"top_p", "frequency_penalty", "presence_penalty", "stop", "seed",
):
if data.get(_k) is not None:
sampling_params[_k] = data[_k]
# OpenAI rejects sending both; the provider maps max_tokens ->
# max_completion_tokens, so drop the alias when the canonical key is present.
if "max_completion_tokens" in sampling_params:
sampling_params.pop("max_tokens", None)
# Check for continuation (tool results after assistant tool_calls)
if is_continuation(messages):
@@ -145,24 +258,41 @@ def translate_request(
"conversation_id": conversation_id,
"tool_actions": tool_actions,
"api_key": api_key,
# Full messages array for STATELESS continuation: OpenAI clients
# (opencode, etc.) don't carry a conversation_id, so the agent is
# rebuilt from the resent messages instead of server-side state.
"messages": messages,
}
# A continuation only exists if turn 1 was saved, so default to True —
# otherwise the final turn and its WAL row are never persisted. An
# explicit override is honoured if the client resends it.
# Stateful continuations (with a conversation_id) default to True so the
# final turn / WAL row is persisted. Stateless continuations (no
# conversation_id, e.g. opencode) default to False — otherwise every
# tool round persists an orphan conversation with an empty question. An
# explicit ``docsgpt.save_conversation`` override is honoured either way.
result["save_conversation"] = bool(
data.get("docsgpt", {}).get("save_conversation", True)
data.get("docsgpt", {}).get("save_conversation", bool(conversation_id))
)
# Carry tools forward for next iteration
if data.get("tools"):
result["client_tools"] = data["tools"]
if response_schema is not None:
result["json_schema"] = response_schema
result["json_schema_strict"] = json_schema_strict
if json_object_mode:
result["json_object"] = True
if sampling_params:
result["llm_params"] = sampling_params
return result
# Normal request — extract question from last user message
question = ""
# Normal request — extract the question (text) from the last user message,
# and keep its full content array (text + image_url parts) when multimodal so
# images still reach the model in the final user message.
last_user_content = None
for msg in reversed(messages):
if msg.get("role") == "user":
question = msg.get("content", "")
last_user_content = msg.get("content")
break
question = content_to_text(last_user_content)
multimodal_content = last_user_content if isinstance(last_user_content, list) else None
history = convert_history(messages)
system_prompt_override = extract_system_prompt(messages)
@@ -189,6 +319,16 @@ def translate_request(
if docsgpt.get("attachments"):
result["attachments"] = docsgpt["attachments"]
if response_schema is not None:
result["json_schema"] = response_schema
result["json_schema_strict"] = json_schema_strict
if json_object_mode:
result["json_object"] = True
if sampling_params:
result["llm_params"] = sampling_params
if multimodal_content is not None:
result["multimodal_content"] = multimodal_content
return result
@@ -205,6 +345,7 @@ def translate_response(
thought: str,
model_name: str,
pending_tool_calls: Optional[List[Dict]] = None,
strip_reasoning_leak: bool = False,
) -> Dict[str, Any]:
"""Translate DocsGPT response to chat completions format.
@@ -246,9 +387,14 @@ def translate_response(
]
finish_reason = "tool_calls"
else:
message["content"] = answer
if thought:
message["reasoning_content"] = thought
if strip_reasoning_leak:
clean_answer, leaked_reasoning = _split_leaked_reasoning(answer)
else:
clean_answer, leaked_reasoning = answer, ""
message["content"] = clean_answer
combined_reasoning = (thought or "") + leaked_reasoning
if combined_reasoning:
message["reasoning_content"] = combined_reasoning
finish_reason = "stop"
result: Dict[str, Any] = {
@@ -312,15 +458,31 @@ def _make_chunk(
return f"data: {json.dumps(chunk)}\n\n"
def _make_docsgpt_chunk(data: Dict[str, Any]) -> str:
"""Build a DocsGPT extension SSE chunk."""
return f"data: {json.dumps({'docsgpt': data})}\n\n"
def _make_docsgpt_chunk(data: Dict[str, Any], completion_id: str, model_name: str) -> str:
"""Build a DocsGPT extension chunk that is ALSO a valid ``chat.completion.chunk``.
Strict OpenAI clients (e.g. the Vercel AI SDK used by opencode) validate every
SSE ``data:`` frame as a chat.completion.chunk, so the DocsGPT extension is
attached to an otherwise-empty (no-op) chunk rather than sent as a bare
``{"docsgpt": ...}`` object — which has no ``choices`` and fails validation.
OpenAI clients ignore the extra top-level ``docsgpt`` field.
"""
chunk = {
"id": completion_id,
"object": "chat.completion.chunk",
"created": int(time.time()),
"model": model_name,
"choices": [{"index": 0, "delta": {}, "finish_reason": None}],
"docsgpt": data,
}
return f"data: {json.dumps(chunk)}\n\n"
def translate_stream_event(
event_data: Dict[str, Any],
completion_id: str,
model_name: str,
strip_reasoning_leak: bool = False,
) -> List[str]:
"""Translate a DocsGPT SSE event dict to standard streaming chunks.
@@ -341,9 +503,18 @@ def translate_stream_event(
chunks: List[str] = []
if event_type == "answer":
chunks.append(
_make_chunk(completion_id, model_name, {"content": event_data.get("answer", "")})
raw = event_data.get("answer", "")
clean, leaked = (
_split_leaked_reasoning(raw) if strip_reasoning_leak else (raw, "")
)
if leaked:
chunks.append(
_make_chunk(completion_id, model_name, {"reasoning_content": leaked})
)
if clean:
chunks.append(
_make_chunk(completion_id, model_name, {"content": clean})
)
elif event_type == "thought":
chunks.append(
@@ -355,10 +526,10 @@ def translate_stream_event(
elif event_type == "source":
chunks.append(
_make_docsgpt_chunk({
"type": "source",
"sources": event_data.get("source", []),
})
_make_docsgpt_chunk(
{"type": "source", "sources": event_data.get("source", [])},
completion_id, model_name,
)
)
elif event_type == "tool_call":
@@ -384,10 +555,10 @@ def translate_stream_event(
)
elif status == "awaiting_approval":
# Extension: approval needed
chunks.append(_make_docsgpt_chunk({"type": "tool_call", "data": tc_data}))
chunks.append(_make_docsgpt_chunk({"type": "tool_call", "data": tc_data}, completion_id, model_name))
elif status in ("completed", "pending", "error", "denied", "skipped"):
# Extension: tool call progress
chunks.append(_make_docsgpt_chunk({"type": "tool_call", "data": tc_data}))
chunks.append(_make_docsgpt_chunk({"type": "tool_call", "data": tc_data}, completion_id, model_name))
elif event_type == "tool_calls_pending":
# Standard: finish_reason = tool_calls
@@ -396,10 +567,13 @@ def translate_stream_event(
)
# Also emit as docsgpt extension
chunks.append(
_make_docsgpt_chunk({
"type": "tool_calls_pending",
"pending_tool_calls": event_data.get("data", {}).get("pending_tool_calls", []),
})
_make_docsgpt_chunk(
{
"type": "tool_calls_pending",
"pending_tool_calls": event_data.get("data", {}).get("pending_tool_calls", []),
},
completion_id, model_name,
)
)
elif event_type == "end":
@@ -409,12 +583,16 @@ def translate_stream_event(
chunks.append("data: [DONE]\n\n")
elif event_type == "id":
chunks.append(
_make_docsgpt_chunk({
"type": "id",
"conversation_id": event_data.get("id", ""),
})
)
# Skip the "None" placeholder conversation_id emitted when the call is
# not persisted (save_conversation=false) — nothing useful to surface.
conv_id = event_data.get("id", "")
if conv_id and conv_id != "None":
chunks.append(
_make_docsgpt_chunk(
{"type": "id", "conversation_id": conv_id},
completion_id, model_name,
)
)
elif event_type == "error":
# Emit as standard error (non-standard but widely supported)
@@ -427,12 +605,18 @@ def translate_stream_event(
chunks.append(f"data: {json.dumps(error_data)}\n\n")
elif event_type == "structured_answer":
chunks.append(
_make_chunk(
completion_id, model_name,
{"content": event_data.get("answer", "")},
)
raw = event_data.get("answer", "")
clean, leaked = (
_split_leaked_reasoning(raw) if strip_reasoning_leak else (raw, "")
)
if leaked:
chunks.append(
_make_chunk(completion_id, model_name, {"reasoning_content": leaked})
)
if clean:
chunks.append(
_make_chunk(completion_id, model_name, {"content": clean})
)
# Skip: tool_calls (redundant), research_plan, research_progress
application/llm/google_ai.py
+6 -2
View File
@@ -653,8 +653,12 @@ class GoogleLLM(BaseLLM):
"""Return whether this LLM supports structured JSON output."""
return True
def prepare_structured_output_format(self, json_schema):
"""Convert JSON schema to Google AI structured output format."""
def prepare_structured_output_format(self, json_schema, strict=True):
"""Convert JSON schema to Google AI structured output format.
``strict`` is accepted for signature parity with the OpenAI provider;
Google enforces the schema natively via ``response_schema``.
"""
if not json_schema:
return None
type_map = {
application/llm/openai.py
+8 -3
View File
@@ -879,7 +879,7 @@ class OpenAILLM(BaseLLM):
if effort:
kwargs["reasoning_effort"] = effort
def prepare_structured_output_format(self, json_schema):
def prepare_structured_output_format(self, json_schema, strict=True):
if not json_schema:
return None
try:
@@ -914,7 +914,12 @@ class OpenAILLM(BaseLLM):
return schema_copy
return schema_obj
processed_schema = add_additional_properties_false(json_schema)
# Strict mode requires additionalProperties:false + all-required on every
# object (OpenAI Structured Outputs). When strict is false (OpenAI's
# lenient json_schema), pass the schema through unchanged.
processed_schema = (
add_additional_properties_false(json_schema) if strict else json_schema
)
result = {
"type": "json_schema",
@@ -924,7 +929,7 @@ class OpenAILLM(BaseLLM):
"description", "Structured response"
),
"schema": processed_schema,
"strict": True,
"strict": strict,
},
}
tests/api/answer/services/test_stream_processor.py
+137
View File
@@ -177,5 +177,142 @@ class TestToolPreFetch:
"""Tests for tool pre-fetching with saved parameter values."""
@pytest.mark.unit
class TestBuildContinuationFromMessages:
"""Stateless tool continuation rebuilt from the resent messages array.
OpenAI-compatible clients (opencode, etc.) resend the full conversation but
carry no conversation_id, so the agent + pending tool calls are reconstructed
directly from the request instead of from server-side ``pending_tool_state``.
"""
@staticmethod
def _make_processor():
from unittest.mock import MagicMock
from application.api.answer.services.stream_processor import StreamProcessor
processor = StreamProcessor({"question": ""}, {"sub": "user_123"})
fake_agent = MagicMock()
fake_agent.tool_executor.get_tools.return_value = {"search": object()}
# build_agent touches the DB / config; stub it for this unit test.
processor.build_agent = MagicMock(return_value=fake_agent)
return processor, fake_agent
def _messages_with_tool_call(self, arguments):
return [
{"role": "system", "content": "You are a bot"},
{"role": "user", "content": "search the docs"},
{
"role": "assistant",
"content": None,
"tool_calls": [
{
"id": "call_1",
"type": "function",
"function": {"name": "search", "arguments": arguments},
}
],
},
{"role": "tool", "tool_call_id": "call_1", "content": "found it"},
]
def test_rebuilds_pending_tool_calls_and_prior_messages(self):
processor, fake_agent = self._make_processor()
messages = self._messages_with_tool_call('{"q": "docs"}')
tool_actions = [{"call_id": "call_1", "result": "found it"}]
(
agent,
prior_messages,
tools_dict,
pending_tool_calls,
returned_actions,
reasoning,
) = processor.build_continuation_from_messages(messages, tool_actions)
assert agent is fake_agent
assert tools_dict == fake_agent.tool_executor.get_tools.return_value
assert returned_actions is tool_actions
assert reasoning == ""
# prior_messages stop before the assistant-with-tool_calls message.
assert prior_messages == messages[:2]
assert len(pending_tool_calls) == 1
call = pending_tool_calls[0]
assert call["call_id"] == "call_1"
assert call["name"] == "search"
assert call["tool_name"] == "search"
assert call["action_name"] == "search"
assert call["arguments"] == {"q": "docs"}
processor.build_agent.assert_called_once_with("")
def test_invalid_json_arguments_default_to_empty_dict(self):
processor, _ = self._make_processor()
messages = self._messages_with_tool_call("not-json")
_, _, _, pending_tool_calls, _, _ = (
processor.build_continuation_from_messages(messages, [])
)
assert pending_tool_calls[0]["arguments"] == {}
def test_dict_arguments_passed_through(self):
processor, _ = self._make_processor()
messages = self._messages_with_tool_call({"already": "parsed"})
_, _, _, pending_tool_calls, _, _ = (
processor.build_continuation_from_messages(messages, [])
)
assert pending_tool_calls[0]["arguments"] == {"already": "parsed"}
def test_uses_last_assistant_tool_call_message(self):
processor, _ = self._make_processor()
messages = [
{"role": "user", "content": "first"},
{
"role": "assistant",
"content": None,
"tool_calls": [
{
"id": "old",
"type": "function",
"function": {"name": "search", "arguments": "{}"},
}
],
},
{"role": "tool", "tool_call_id": "old", "content": "r1"},
{"role": "user", "content": "second"},
{
"role": "assistant",
"content": None,
"tool_calls": [
{
"id": "new",
"type": "function",
"function": {"name": "search", "arguments": "{}"},
}
],
},
{"role": "tool", "tool_call_id": "new", "content": "r2"},
]
_, prior_messages, _, pending_tool_calls, _, _ = (
processor.build_continuation_from_messages(messages, [])
)
assert pending_tool_calls[0]["call_id"] == "new"
# Everything up to (not including) the last assistant tool_calls message.
assert prior_messages == messages[:4]
def test_raises_when_no_assistant_tool_calls(self):
processor, _ = self._make_processor()
messages = [
{"role": "user", "content": "hi"},
{"role": "assistant", "content": "no tools here"},
]
with pytest.raises(ValueError, match="No assistant message with tool_calls"):
processor.build_continuation_from_messages(messages, [])
tests/api/v1/test_routes_extended.py
+3 -1
View File
@@ -541,7 +541,9 @@ class TestChatCompletionsHappyPath:
translated_chunks: list = []
def _fake_translate_stream_event(event_data, completion_id, model_name):
def _fake_translate_stream_event(
event_data, completion_id, model_name, strip_reasoning_leak=False
):
translated_chunks.append(event_data)
return ['data: x\n\n']
tests/test_v1_translator.py
+419 -7
View File
@@ -10,7 +10,11 @@ import pytest
from application.api.v1.translator import (
_get_client_tool_name,
_split_leaked_reasoning,
_strip_repr_quotes,
content_to_text,
convert_history,
extract_response_schema,
extract_system_prompt,
extract_tool_results,
is_continuation,
@@ -245,10 +249,12 @@ class TestTranslateRequest:
assert len(result["tool_actions"]) == 1
assert result["tool_actions"][0]["call_id"] == "c1"
def test_continuation_persists_by_default(self):
"""A continuation implies the first turn was saved, so the resumed turn
must persist too (otherwise the final answer + WAL row are lost)."""
def test_stateful_continuation_persists_by_default(self):
"""A stateful continuation (conversation_id present) implies the first
turn was saved, so the resumed turn must persist too (otherwise the
final answer + WAL row are lost)."""
data = {
"conversation_id": "conv-1",
"messages": [
{"role": "user", "content": "Search for X"},
{
@@ -261,6 +267,23 @@ class TestTranslateRequest:
result = translate_request(data, "key")
assert result["save_conversation"] is True
def test_stateless_continuation_does_not_persist_by_default(self):
"""A stateless continuation (no conversation_id, e.g. an OpenAI client
such as opencode) never persisted turn 1, so it must NOT default to
saving -- otherwise every tool round leaks an orphan conversation."""
data = {
"messages": [
{"role": "user", "content": "Search for X"},
{
"role": "assistant",
"tool_calls": [{"id": "c1", "type": "function", "function": {"name": "search", "arguments": "{}"}}],
},
{"role": "tool", "tool_call_id": "c1", "content": "done"},
],
}
result = translate_request(data, "key")
assert result["save_conversation"] is False
def test_continuation_honours_explicit_save_conversation_override(self):
"""An explicit docsgpt.save_conversation=false on the continuation wins."""
data = {
@@ -650,15 +673,23 @@ class TestTranslateStreamEvent:
assert parsed["docsgpt"]["type"] == "tool_call"
def test_standard_clients_can_ignore_docsgpt(self):
"""Standard clients parse only 'choices' — docsgpt namespace is ignored."""
"""The docsgpt namespace rides on a valid empty chat.completion.chunk.
Standard OpenAI clients validate every streamed frame as a
``chat.completion.chunk``; a frame without ``choices`` is rejected.
So docsgpt-only events (sources, ids, tool_calls) are emitted as a
valid chunk with an empty delta plus a ``docsgpt`` extension that
standard parsers simply ignore.
"""
chunks = translate_stream_event(
{"type": "source", "source": [{"title": "t"}]},
"chatcmpl-1", "agent",
)
parsed = json.loads(chunks[0].replace("data: ", "").strip())
# No "choices" key — standard parsers skip this chunk entirely
assert "choices" not in parsed
# docsgpt key is present
# Valid chunk envelope: standard parsers read choices[0].delta (empty)
assert isinstance(parsed.get("choices"), list)
assert parsed["choices"][0]["delta"] == {}
# docsgpt extension is present for clients that want it
assert "docsgpt" in parsed
@@ -681,3 +712,384 @@ class TestGetClientToolName:
def test_returns_empty_when_no_fields(self):
assert _get_client_tool_name({}) == ""
# ---------------------------------------------------------------------------
# content_to_text
# ---------------------------------------------------------------------------
@pytest.mark.unit
class TestContentToText:
def test_plain_string_passthrough(self):
assert content_to_text("hello") == "hello"
def test_none_returns_empty(self):
assert content_to_text(None) == ""
def test_text_parts_joined(self):
content = [
{"type": "text", "text": "line one"},
{"type": "text", "text": "line two"},
]
assert content_to_text(content) == "line one\nline two"
def test_image_parts_dropped(self):
content = [
{"type": "text", "text": "describe this"},
{"type": "image_url", "image_url": {"url": "data:image/png;base64,xxx"}},
]
# Only the text part contributes; the image is dropped from the flattened text.
assert content_to_text(content) == "describe this"
def test_bare_string_parts_included(self):
assert content_to_text(["a", "b"]) == "a\nb"
def test_missing_text_field_becomes_empty(self):
assert content_to_text([{"type": "text"}]) == ""
def test_non_string_non_list_coerced(self):
assert content_to_text(123) == "123"
# ---------------------------------------------------------------------------
# extract_response_schema
# ---------------------------------------------------------------------------
_SCHEMA = {
"type": "object",
"properties": {"answer": {"type": "string"}},
"required": ["answer"],
}
@pytest.mark.unit
class TestExtractResponseSchema:
def test_none_when_absent(self):
assert extract_response_schema({}) is None
def test_response_format_json_schema_wrapper(self):
data = {
"response_format": {
"type": "json_schema",
"json_schema": {"name": "ans", "schema": _SCHEMA},
}
}
assert extract_response_schema(data) == _SCHEMA
def test_response_format_bare_schema_under_json_schema(self):
# A bare schema (no "schema" wrapper) but with a top-level "type" is tolerated.
data = {
"response_format": {
"type": "json_schema",
"json_schema": _SCHEMA,
}
}
assert extract_response_schema(data) == _SCHEMA
def test_response_format_json_object_carries_no_schema(self):
data = {"response_format": {"type": "json_object"}}
assert extract_response_schema(data) is None
def test_response_schema_raw_object(self):
assert extract_response_schema({"response_schema": _SCHEMA}) == _SCHEMA
def test_response_schema_wrapper(self):
data = {"response_schema": {"schema": _SCHEMA}}
assert extract_response_schema(data) == _SCHEMA
def test_response_schema_takes_precedence_over_response_format(self):
other = {"type": "object", "properties": {}}
data = {
"response_schema": _SCHEMA,
"response_format": {
"type": "json_schema",
"json_schema": {"schema": other},
},
}
assert extract_response_schema(data) == _SCHEMA
# ---------------------------------------------------------------------------
# _split_leaked_reasoning
# ---------------------------------------------------------------------------
@pytest.mark.unit
class TestSplitLeakedReasoning:
def test_no_marker_is_noop(self):
assert _split_leaked_reasoning("just an answer") == ("just an answer", "")
def test_none_is_noop(self):
assert _split_leaked_reasoning(None) == (None, "")
def test_extracts_single_thought(self):
content = "{'type': 'thought', 'thought': 'let me think'}The answer is 42."
clean, leaked = _split_leaked_reasoning(content)
assert clean == "The answer is 42."
assert leaked == "let me think"
def test_extracts_multiple_thoughts(self):
content = (
"{'type': 'thought', 'thought': 'first'}"
"partial"
"{'type': 'thought', 'thought': 'second'}done"
)
clean, leaked = _split_leaked_reasoning(content)
assert clean == "partialdone"
assert leaked == "firstsecond"
def test_double_quoted_thought_value(self):
# When the token contains an apostrophe the repr uses double quotes.
content = "{'type': 'thought', 'thought': \"I'll check\"}answer"
clean, leaked = _split_leaked_reasoning(content)
assert clean == "answer"
assert leaked == "I'll check"
def test_thought_value_with_brace_not_truncated(self):
content = "{'type': 'thought', 'thought': 'use {json} here'}final"
clean, leaked = _split_leaked_reasoning(content)
assert clean == "final"
assert leaked == "use {json} here"
def test_strip_repr_quotes_unquoted_passthrough(self):
# Defensive branch: an unquoted value is returned unchanged.
assert _strip_repr_quotes("plain") == "plain"
assert _strip_repr_quotes("'quoted'") == "quoted"
# ---------------------------------------------------------------------------
# translate_request — structured outputs / sampling / multimodal
# ---------------------------------------------------------------------------
@pytest.mark.unit
class TestTranslateRequestStructuredOutputs:
def test_response_format_surfaces_json_schema_strict_default(self):
data = {
"messages": [{"role": "user", "content": "Hi"}],
"response_format": {
"type": "json_schema",
"json_schema": {"name": "ans", "schema": _SCHEMA},
},
}
result = translate_request(data, "key")
assert result["json_schema"] == _SCHEMA
assert result["json_schema_strict"] is True
def test_response_format_honours_explicit_strict_false(self):
data = {
"messages": [{"role": "user", "content": "Hi"}],
"response_format": {
"type": "json_schema",
"json_schema": {"name": "ans", "schema": _SCHEMA, "strict": False},
},
}
result = translate_request(data, "key")
assert result["json_schema_strict"] is False
def test_json_object_mode_flag(self):
data = {
"messages": [{"role": "user", "content": "Hi"}],
"response_format": {"type": "json_object"},
}
result = translate_request(data, "key")
assert result["json_object"] is True
assert "json_schema" not in result
def test_sampling_params_forwarded(self):
data = {
"messages": [{"role": "user", "content": "Hi"}],
"temperature": 0.2,
"top_p": 0.9,
"seed": 7,
}
result = translate_request(data, "key")
assert result["llm_params"] == {"temperature": 0.2, "top_p": 0.9, "seed": 7}
def test_max_tokens_alias_dropped_when_canonical_present(self):
data = {
"messages": [{"role": "user", "content": "Hi"}],
"max_tokens": 100,
"max_completion_tokens": 200,
}
result = translate_request(data, "key")
assert result["llm_params"]["max_completion_tokens"] == 200
assert "max_tokens" not in result["llm_params"]
def test_multimodal_content_preserved_and_question_flattened(self):
content = [
{"type": "text", "text": "what is this"},
{"type": "image_url", "image_url": {"url": "data:image/png;base64,xxx"}},
]
data = {"messages": [{"role": "user", "content": content}]}
result = translate_request(data, "key")
assert result["question"] == "what is this"
assert result["multimodal_content"] == content
def test_plain_text_request_has_no_multimodal_content(self):
data = {"messages": [{"role": "user", "content": "plain"}]}
result = translate_request(data, "key")
assert "multimodal_content" not in result
def test_continuation_forwards_schema_and_sampling(self):
data = {
"messages": [
{"role": "user", "content": "Hi"},
{
"role": "assistant",
"content": None,
"tool_calls": [
{
"id": "call_1",
"type": "function",
"function": {"name": "search", "arguments": "{}"},
}
],
},
{"role": "tool", "tool_call_id": "call_1", "content": "result"},
],
"response_format": {
"type": "json_schema",
"json_schema": {"name": "ans", "schema": _SCHEMA},
},
"temperature": 0.5,
}
result = translate_request(data, "key")
assert result["tool_actions"]
assert result["json_schema"] == _SCHEMA
assert result["json_schema_strict"] is True
assert result["llm_params"] == {"temperature": 0.5}
def test_continuation_forwards_tools_and_json_object(self):
data = {
"messages": [
{"role": "user", "content": "Hi"},
{
"role": "assistant",
"content": None,
"tool_calls": [
{
"id": "call_1",
"type": "function",
"function": {"name": "search", "arguments": "{}"},
}
],
},
{"role": "tool", "tool_call_id": "call_1", "content": "result"},
],
"tools": [{"type": "function", "function": {"name": "search"}}],
"response_format": {"type": "json_object"},
}
result = translate_request(data, "key")
assert result["client_tools"] == data["tools"]
assert result["json_object"] is True
# ---------------------------------------------------------------------------
# translate_response / translate_stream_event — reasoning-leak handling
# ---------------------------------------------------------------------------
@pytest.mark.unit
class TestReasoningLeakHandling:
def test_response_strips_leak_only_when_enabled(self):
answer = "{'type': 'thought', 'thought': 'hmm'}Final answer."
result = translate_response(
conversation_id="",
answer=answer,
sources=None,
tool_calls=None,
thought="",
model_name="agent",
strip_reasoning_leak=True,
)
msg = result["choices"][0]["message"]
assert msg["content"] == "Final answer."
assert msg["reasoning_content"] == "hmm"
def test_response_preserves_content_when_strip_disabled(self):
answer = "{'type': 'thought', 'thought': 'hmm'}Final answer."
result = translate_response(
conversation_id="",
answer=answer,
sources=None,
tool_calls=None,
thought="",
model_name="agent",
strip_reasoning_leak=False,
)
msg = result["choices"][0]["message"]
# Untouched: the leak marker is left in content verbatim.
assert msg["content"] == answer
assert "reasoning_content" not in msg
def test_response_combines_thought_and_leaked_reasoning(self):
answer = "{'type': 'thought', 'thought': 'leaked'}Answer."
result = translate_response(
conversation_id="",
answer=answer,
sources=None,
tool_calls=None,
thought="explicit ",
model_name="agent",
strip_reasoning_leak=True,
)
msg = result["choices"][0]["message"]
assert msg["reasoning_content"] == "explicit leaked"
def test_stream_answer_splits_leak_when_enabled(self):
chunks = translate_stream_event(
{"type": "answer", "answer": "{'type': 'thought', 'thought': 'r'}hi"},
"chatcmpl-1", "agent", True,
)
deltas = [
json.loads(c.replace("data: ", "").strip())["choices"][0]["delta"]
for c in chunks
]
assert {"reasoning_content": "r"} in deltas
assert {"content": "hi"} in deltas
def test_stream_answer_preserves_content_when_disabled(self):
raw = "{'type': 'thought', 'thought': 'r'}hi"
chunks = translate_stream_event(
{"type": "answer", "answer": raw}, "chatcmpl-1", "agent", False,
)
deltas = [
json.loads(c.replace("data: ", "").strip())["choices"][0]["delta"]
for c in chunks
]
assert {"content": raw} in deltas
assert all("reasoning_content" not in d for d in deltas)
def test_stream_structured_answer_event(self):
chunks = translate_stream_event(
{"type": "structured_answer", "answer": '{"answer": "42"}'},
"chatcmpl-1", "agent", True,
)
deltas = [
json.loads(c.replace("data: ", "").strip())["choices"][0]["delta"]
for c in chunks
]
assert {"content": '{"answer": "42"}'} in deltas
def test_stream_structured_answer_splits_leak(self):
chunks = translate_stream_event(
{
"type": "structured_answer",
"answer": "{'type': 'thought', 'thought': 'why'}{\"answer\": \"42\"}",
},
"chatcmpl-1", "agent", True,
)
deltas = [
json.loads(c.replace("data: ", "").strip())["choices"][0]["delta"]
for c in chunks
]
assert {"reasoning_content": "why"} in deltas
assert {"content": '{"answer": "42"}'} in deltas