15254528833

Committed 26 May 2025 12:56PM UTC coverage: 90.146% (-0.3%) from 90.411%

Build # 15254528833

Build Type

Pull #9426

github

Committed by

web-flow

Commit Message

Merge 06c2b66b1 into 802328e29

Pull Request Pull Request #9426: feat: add component name and type to `StreamingChunk`

Run Details

11398 of 12644 relevant lines covered (90.15%)

0.9 hits per line

Source File
Press 'n' to go to next uncovered line, 'b' for previous

20.0

haystack/components/generators/utils.py

# SPDX-FileCopyrightText: 2022-present deepset GmbH <info@deepset.ai>
#
# SPDX-License-Identifier: Apache-2.0

from typing import Any, Dict

from openai.types.chat.chat_completion_chunk import ChoiceDeltaToolCall

from haystack.dataclasses import StreamingChunk


def print_streaming_chunk(chunk: StreamingChunk) -> None:
    """
    Callback function to handle and display streaming output chunks.

    This function processes a `StreamingChunk` object by:
    - Printing tool call metadata (if any), including function names and arguments, as they arrive.
    - Printing tool call results when available.
    - Printing the main content (e.g., text tokens) of the chunk as it is received.

    The function outputs data directly to stdout and flushes output buffers to ensure immediate display during
    streaming.

    :param chunk: A chunk of streaming data containing content and optional metadata, such as tool calls and
        tool results.
    """
    # Print tool call metadata if available (from ChatGenerator)
    if tool_calls := chunk.meta.get("tool_calls"):
        for tool_call in tool_calls:
            # Convert to dict if tool_call is a ChoiceDeltaToolCall
            tool_call_dict: Dict[str, Any] = (
                tool_call.to_dict() if isinstance(tool_call, ChoiceDeltaToolCall) else tool_call
            )

            if function := tool_call_dict.get("function"):
                if name := function.get("name"):
                    print("\n\n[TOOL CALL]\n", flush=True, end="")
                    print(f"Tool: {name} ", flush=True, end="")
                    print("\nArguments: ", flush=True, end="")

                if arguments := function.get("arguments"):
                    print(arguments, flush=True, end="")

    # Print tool call results if available (from ToolInvoker)
    if tool_result := chunk.meta.get("tool_result"):
        print(f"\n\n[TOOL RESULT]\n{tool_result}", flush=True, end="")

    # Print the main content of the chunk (from ChatGenerator)
    if content := chunk.content:
        print(content, flush=True, end="")

    # End of LLM assistant message so we add two new lines
    # This ensures spacing between multiple LLM messages (e.g. Agent)
    if chunk.meta.get("finish_reason") is not None:
        print("\n\n", flush=True, end="")

1	# SPDX-FileCopyrightText: 2022-present deepset GmbH <info@deepset.ai>
2	#
3	# SPDX-License-Identifier: Apache-2.0
4
5	from typing import Any, Dict	1✔
6
7	from openai.types.chat.chat_completion_chunk import ChoiceDeltaToolCall	1✔
8
9	from haystack.dataclasses import StreamingChunk	1✔
10
11
12	def print_streaming_chunk(chunk: StreamingChunk) -> None:	1✔
13	"""
14	Callback function to handle and display streaming output chunks.
15
16	This function processes a `StreamingChunk` object by:
17	- Printing tool call metadata (if any), including function names and arguments, as they arrive.
18	- Printing tool call results when available.
19	- Printing the main content (e.g., text tokens) of the chunk as it is received.
20
21	The function outputs data directly to stdout and flushes output buffers to ensure immediate display during
22	streaming.
23
24	:param chunk: A chunk of streaming data containing content and optional metadata, such as tool calls and
25	tool results.
26	"""
27	# Print tool call metadata if available (from ChatGenerator)
28	if tool_calls := chunk.meta.get("tool_calls"):	×
29	for tool_call in tool_calls:	×
30	# Convert to dict if tool_call is a ChoiceDeltaToolCall
31	tool_call_dict: Dict[str, Any] = (	×
32	tool_call.to_dict() if isinstance(tool_call, ChoiceDeltaToolCall) else tool_call
33	)
34
35	if function := tool_call_dict.get("function"):	×
36	if name := function.get("name"):	×
37	print("\n\n[TOOL CALL]\n", flush=True, end="")	×
38	print(f"Tool: {name} ", flush=True, end="")	×
39	print("\nArguments: ", flush=True, end="")	×
40
41	if arguments := function.get("arguments"):	×
42	print(arguments, flush=True, end="")	×
43
44	# Print tool call results if available (from ToolInvoker)
45	if tool_result := chunk.meta.get("tool_result"):	×
46	print(f"\n\n[TOOL RESULT]\n{tool_result}", flush=True, end="")	×
47
48	# Print the main content of the chunk (from ChatGenerator)
49	if content := chunk.content:	×
50	print(content, flush=True, end="")	×
51
52	# End of LLM assistant message so we add two new lines
53	# This ensures spacing between multiple LLM messages (e.g. Agent)
54	if chunk.meta.get("finish_reason") is not None:	×
55	print("\n\n", flush=True, end="")	×

deepset-ai / haystack / 15254528833

Source File Press 'n' to go to next uncovered line, 'b' for previous

Source File
Press 'n' to go to next uncovered line, 'b' for previous