Skip to content

tools_node

llmfy.flow_engine.helper.tools_node.tools_node

tools_node(messages, registry)

Tools Node

Parameters:

Name Type Description Default
messages List[Message]

List message

 required
registry ToolRegistry

Tool registry

 required

Returns:

Type Description
List[Message]

Tool results message
Source code in llmfy/flow_engine/helper/tools_node/tools_node.py 
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
def tools_node(messages: List[Message], registry: ToolRegistry) -> List[Message]:
    """Tools Node

    Args:
        messages (List[Message]): List message
        registry (ToolRegistry): Tool registry

    Returns:
        Tool results message
    """
    new_messages = deepcopy(messages)
    last_message = new_messages[-1]
    tool_results: List[Message] = []
    if last_message.tool_calls:
        for tool_call in last_message.tool_calls:
            result = registry.execute_tool(
                name=tool_call.name, arguments=tool_call.arguments
            )
            tool_results.append(
                Message(
                    role=Role.TOOL,
                    request_call_id=tool_call.request_call_id,
                    tool_call_id=tool_call.tool_call_id,
                    tool_results=[str(result)],
                    name=tool_call.name,
                )
            )
    return tool_results

tools_stream_node(messages, registry)

Tools Stream Node

Parameters:

Name Type Description Default
messages List[Message]

List message

 required
registry ToolRegistry

Tool registry

 required

Yields:

Type Description
ToolNodeStreamResponse

Tool results message
Source code in llmfy/flow_engine/helper/tools_node/tools_node.py 
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
def tools_stream_node(messages: List[Message], registry: ToolRegistry) -> Generator[ToolNodeStreamResponse, Any, None]:
    """Tools Stream Node

    Args:
        messages (List[Message]): List message
        registry (ToolRegistry): Tool registry

    Yields:
        Tool results message
    """
    new_messages = deepcopy(messages)
    last_message = new_messages[-1]
    response = ToolNodeStreamResponse()

    if last_message.tool_calls:
        for tool_call in last_message.tool_calls:
            # Executing
            response.type = ToolNodeStreamType.EXECUTING
            response.name = tool_call.name
            response.arguments = tool_call.arguments
            response.result = None
            yield response

            result = registry.execute_tool(
                name=tool_call.name, arguments=tool_call.arguments
            )
            tool_result = Message(
                role=Role.TOOL,
                request_call_id=tool_call.request_call_id,
                tool_call_id=tool_call.tool_call_id,
                tool_results=[str(result)],
                name=tool_call.name,
            )

            # Result
            response.type = ToolNodeStreamType.RESULT
            response.name = tool_call.name
            response.arguments = tool_call.arguments
            response.result = tool_result
            yield response