Function execute_tool

src/WorkFlow.py:69–102

(name: str, state: PipelineState, prompt_template: str, llm)

    return state

def execute_tool(name: str, state: PipelineState, prompt_template: str, llm) -> PipelineState:

    logger(f"{name} is working...")

    state["history"] = clip_history(state["history"])
    
    generation = create_llm_chain(prompt_template, llm, state["history"])

    # Sanitize the generation output by removing invalid control characters
    sanitized_generation = re.sub(r'[\x00-\x1F\x7F]', '', generation)

    logger(sanitized_generation)

    data = json.loads(sanitized_generation)
    
    choice = data
    tool_name = choice["function"]
    args = choice["args"]
    
    if tool_name not in tool_registry:
        raise ValueError(f"Tool {tool_name} not found in registry.")
    
    result = tool_registry[tool_name](*args)

    # Flatten args to a string
    flattened_args = ', '.join(map(str, args))

    logger(f"\nExecuted Tool: {tool_name}({flattened_args})  Result is: {result}")


    state["history"] += f"\nExecuted {tool_name}({flattened_args})  Result is: {result}"
    state["history"] = clip_history(state["history"])

    return state

def condition_switch(name:str, state: PipelineState, prompt_template: str, llm) -> PipelineState:
    logger(f"{name} is working...")