Python: Enhance tool call handling and function result processing (#790)

* Tool call enhancement

* add tests

* fix comments

* resolve comments

---------

Co-authored-by: Chris <66376200+crickman@users.noreply.github.com>
Co-authored-by: Eric Zhu <ekzhu@users.noreply.github.com>
This commit is contained in:
Yuge Zhang
2025-09-18 12:13:30 +08:00
committed by GitHub
Unverified
parent 4c80e7017d
commit 1dba779f76
3 changed files with 216 additions and 14 deletions
@@ -185,10 +185,10 @@ class OpenAIBaseChatClient(OpenAIBase, BaseChatClient):
if choice.finish_reason:
finish_reason = FinishReason(value=choice.finish_reason)
contents: list[Contents] = []
if parsed_tool_calls := [tool for tool in self._get_tool_calls_from_chat_choice(choice)]:
contents.extend(parsed_tool_calls)
if text_content := self._parse_text_from_choice(choice):
contents.append(text_content)
if parsed_tool_calls := [tool for tool in self._get_tool_calls_from_chat_choice(choice)]:
contents.extend(parsed_tool_calls)
messages.append(ChatMessage(role="assistant", contents=contents))
return ChatResponse(
response_id=response.id,
@@ -354,8 +354,13 @@ class OpenAIBaseChatClient(OpenAIBase, BaseChatClient):
args["tool_calls"] = [self._openai_content_parser(content)] # type: ignore
case FunctionResultContent():
args["tool_call_id"] = content.call_id
if content.result:
if content.result is not None:
args["content"] = prepare_function_call_results(content.result)
elif content.exception is not None:
# Send the exception message to the model
# Otherwise we won't have any channels to talk to OpenAI
# TODO(yuge): This should ideally be customizable
args["content"] = "Error: " + str(content.exception)
case _:
if "content" not in args:
args["content"] = []
@@ -50,21 +50,28 @@ __all__ = [
]
def _prepare_function_call_results_as_dumpable(content: Contents | Any | list[Contents | Any]) -> Any:
if isinstance(content, list):
# Particularly deal with lists of BaseModel
return [_prepare_function_call_results_as_dumpable(item) for item in content]
if isinstance(content, dict):
return {k: _prepare_function_call_results_as_dumpable(v) for k, v in content.items()}
if isinstance(content, BaseModel):
return content.model_dump(exclude={"raw_representation", "additional_properties"})
return content
def prepare_function_call_results(content: Contents | Any | list[Contents | Any]) -> str | list[str]:
"""Prepare the values of the function call results."""
if isinstance(content, list):
results: list[str] = []
for item in content:
res = prepare_function_call_results(item)
if isinstance(res, list):
results.extend(res)
else:
results.append(res)
return results[0] if len(results) == 1 else json.dumps(results)
if isinstance(content, BaseModel):
return content.model_dump_json(exclude_none=True, exclude={"raw_representation", "additional_properties"})
# BaseModel is already dumpable, shortcut for performance
return content.model_dump_json(exclude={"raw_representation", "additional_properties"})
dumpable = _prepare_function_call_results_as_dumpable(content)
if isinstance(dumpable, str):
return dumpable
# fallback
return json.dumps(content)
return json.dumps(dumpable)
class OpenAISettings(AFBaseSettings):