base.py

Overview

The base.py file defines foundational classes and data structures used for integrating and managing "tools" within the InfiniFlow agent framework. These tools are components that expose functionality (such as API calls or computational logic) which can be invoked programmatically and managed within a canvas-based workflow environment.

Key responsibilities of this file include:

• Defining standard metadata schemas for tools and their parameters.

• Providing base classes for tool parameters (ToolParamBase) and tools themselves (ToolBase).

• Implementing a session class (LLMToolPluginCallSession) to facilitate invoking tools by name with arguments, including timing and callback support.

• Utility methods for retrieving and formatting chunked content from tool responses to support knowledge base generation.

This file plays a central role in the architecture by establishing the interface and lifecycle of tools, ensuring consistency in parameter handling, invocation, and output management, and by bridging between lower-level tool implementations and the higher-level canvas workflow system.

Classes and Functions

TypedDicts for Metadata

ToolParameter

class ToolParameter(TypedDict):
    type: str
    description: str
    displayDescription: str
    enum: List[str]
    required: bool

• Purpose: Defines the structure of a single tool parameter's metadata.

• Attributes:

  • type: The data type of the parameter (e.g., "string", "integer").

  • description: A brief textual description of the parameter.

  • displayDescription: A user-friendly description for UI display.

  • enum: Optional list of allowed string values (enumeration).

  • required: Boolean flag indicating if the parameter is mandatory.

ToolMeta

class ToolMeta(TypedDict):
    name: str
    displayName: str
    description: str
    displayDescription: str
    parameters: dict[str, ToolParameter]

• Purpose: Defines the structure of metadata describing a tool.

• Attributes:

  • name: Unique tool identifier.

  • displayName: Human-readable name.

  • description: Technical description.

  • displayDescription: User-friendly description.

  • parameters: Dictionary mapping parameter names to ToolParameter metadata.

Class: LLMToolPluginCallSession

class LLMToolPluginCallSession(ToolCallSession):

• Purpose: Manages a session to call tools by name, handling both direct invocation and MCP tool sessions.

• Constructor:

  • tools_map: dict[str, object] — Mapping of tool names to tool objects.

  • callback: partial — A callback function to be invoked after each tool call with details.

• Methods:

  • tool_call(name: str, arguments: dict[str, Any]) -> Any
    Invokes the tool identified by name with the provided arguments.

    • If the tool object is an instance of MCPToolCallSession, it calls its tool_call method with a timeout of 60 seconds.

    • Otherwise, it calls the tool's invoke method with the arguments unpacked as keywords.

    • Measures elapsed time and triggers the callback with invocation details.

    • Returns the tool's response.

  • get_tool_obj(name)
    Returns the underlying tool object for the given name.

• Usage example:

tools = {
    "search": search_tool_instance,
    "translate": MCPToolCallSession(...)
}
def callback(name, args, resp, elapsed_time):
    print(f"Tool {name} called in {elapsed_time:.2f}s")
session = LLMToolPluginCallSession(tools, callback=callback)
result = session.tool_call("search", {"query": "example"})

Class: ToolParamBase

class ToolParamBase(ComponentParamBase):

• Purpose: Base class for tool parameter objects, managing tool metadata and default values.

• Initialization:

  • Calls _init_inputs() to deep-copy parameter metadata into self.inputs.

  • Calls _init_attr_by_meta() to create attributes on the instance for each parameter, setting default values if specified.

• Methods:

  • _init_inputs()
    Creates a deep copy of all parameters from the tool's metadata into self.inputs.

  • _init_attr_by_meta()
    For each parameter, sets an instance attribute to the parameter's default value if the attribute doesn't already exist.

  • get_meta() -> dict
    Returns a dictionary describing the tool's function signature in a JSON schema-like format, including:

    • Function name and description (overridable via instance attributes).

    • Parameters' types, descriptions, enumerations, and required flags.

• Implementation notes:

  • Designed to be subclassed where self.meta is set to the tool's metadata.

  • Supports dynamic parameter introspection for UI generation or validation.

Class: ToolBase

class ToolBase(ComponentBase):

• Purpose: Base class for all tool components, managing lifecycle, invocation, and output handling within a canvas.

• Constructor parameters:

  • canvas — Instance of Canvas (imported locally to avoid cyclic dependencies).

  • id — Unique identifier for this tool instance.

  • param: ComponentParamBase — Parameter object, typically subclass of ToolParamBase.

• Initialization:

  • Asserts canvas is an instance of Canvas.

  • Stores parameters and calls param.check() to validate parameters.

• Methods:

  • get_meta() -> dict
    Returns metadata by delegating to the parameter object's get_meta().

  • invoke(**kwargs) -> Any
    Public method to invoke the tool with given parameters.

    • Records creation time.

    • Wraps the internal _invoke() in try-except to catch exceptions, store error info in outputs, and log errors.

    • Clears debug inputs after invocation.

    • Records elapsed time in outputs.

    • Returns the result or error string.

  • _retrieve_chunks(res_list: list, get_title, get_url, get_content, get_score=None)
    Helper to process a list of result objects into "chunks" for knowledge base references.

    • Extracts relevant fields using provided getter functions.

    • Cleans content from embedded base64 images.

    • Truncates content to 10,000 characters.

    • Hashes content to generate stable IDs.

    • Aggregates documents and adds references to the canvas.

    • Sets a formatted content output using the kb_prompt template.

  • thoughts() -> str
    Returns a string indicating the tool is currently running, combining its component name and status.

• Usage example:

class MyToolParam(ToolParamBase):
    meta = {
        "name": "my_tool",
        "description": "Example tool",
        "parameters": {
            "param1": {"type": "string", "description": "Input param", "required": True}
        }
    }

class MyTool(ToolBase):
    def _invoke(self, param1):
        # Tool logic here
        return f"Processed {param1}"

canvas = Canvas()
param = MyToolParam()
tool = MyTool(canvas, "tool1", param)
result = tool.invoke(param1="test")
print(result)  # Output: Processed test

Important Implementation Details

• Deferred import of Canvas:
  To avoid circular dependencies, the Canvas class is imported inside the ToolBase constructor rather than at the module level.

• Timing and error handling in invoke():
  The invoke method measures execution time using time.perf_counter() and captures exceptions to provide robust error reporting and logging.

• Content chunking in _retrieve_chunks():
  This method cleans out embedded base64 images from content, truncates large content, hashes it for stable identifiers, and organizes results for knowledge base integration. The generated chunks and aggregates are added as references to the canvas, enabling further processing or display.

• Use of TypedDict for structured metadata:
  The use of TypedDict for ToolParameter and ToolMeta provides explicit type hints and enforces consistent metadata structure, useful for validation and tooling.

• Callback mechanism in LLMToolPluginCallSession:
  Enables external monitoring or logging of tool calls with elapsed time measurements.

Interaction with Other System Components

• Canvas (agent.canvas.Canvas):
  Every tool is associated with a canvas instance, which manages the workflow and references. The tool adds references to the canvas and queries it for component names.

• Component base classes (agent.component.base.ComponentParamBase, ComponentBase):
  ToolParamBase and ToolBase extend these foundational classes, inheriting parameter validation and component lifecycle behavior.

• MCPToolCallSession (rag.utils.mcp_tool_call_conn.MCPToolCallSession):
  Provides a specialized tool session used in LLMToolPluginCallSession for tools that require multi-call processing.

• Prompt templates (rag.prompts.prompts.kb_prompt):
  Used to format chunked content for knowledge base integration.

• API utility (api.utils.hash_str2int):
  Generates stable integer hashes from string content for chunk/document IDs.

• RAG LLM Chat Model (rag.llm.chat_model.ToolCallSession):
  LLMToolPluginCallSession subclasses this, integrating with the retrieval-augmented generation framework.

Diagram: Class Structure and Relationships

classDiagram
    class ToolParameter {
        +type: str
        +description: str
        +displayDescription: str
        +enum: List[str]
        +required: bool
    }

    class ToolMeta {
        +name: str
        +displayName: str
        +description: str
        +displayDescription: str
        +parameters: dict[str, ToolParameter]
    }

    class LLMToolPluginCallSession {
        -tools_map: dict[str, object]
        -callback: partial
        +__init__(tools_map, callback)
        +tool_call(name: str, arguments: dict) Any
        +get_tool_obj(name) object
    }

    class ToolParamBase {
        -inputs: dict
        +__init__()
        +_init_inputs()
        +_init_attr_by_meta()
        +get_meta() dict
    }

    class ToolBase {
        -_canvas: Canvas
        -_id: str
        -_param: ComponentParamBase
        +__init__(canvas, id, param)
        +get_meta() dict
        +invoke(**kwargs) Any
        -_retrieve_chunks(res_list, get_title, get_url, get_content, get_score=None)
        +thoughts() str
    }

    ToolParamBase --|> ComponentParamBase
    ToolBase --|> ComponentBase
    LLMToolPluginCallSession --|> ToolCallSession

Summary

The base.py file provides the core abstractions and utilities for defining, parameterizing, invoking, and managing tools in the InfiniFlow agent platform. It enables consistent metadata handling, robust invocation with error and time tracking, and supports integration into the canvas-driven workflow system. The LLMToolPluginCallSession facilitates tool invocation in an LLM context, bridging between tool implementations and conversational agents.

This base layer is essential for extending the system with new tools and ensuring they adhere to expected interfaces and behaviors within the agent ecosystem.