Arklex AI

Arklex is an agent-first framework designed to bridge the gap between structured AI workflows and autonomous AI agents. It combines the robustness of structured workflows with the adaptability of modern agents, making it suitable for complex applications like AI-driven coding, research, and automation.

Core Components

• Task Graphs: Structured graphs of agents, workers, and tools that encode reusable logic and workflows.
• Agents: LLM-powered planners that break down complex tasks and make tool and worker calls through step-by-step reasoning.
• Workers: Executable units (or subgraphs) that perform tasks or recursively invoke other workers.
• Tools: Functional units (API calls, search, retrieval, etc.) with built-in validation.
• Natural Language Understanding (NLU): Semantic parsing of user input for intent and argument extraction.
• Task Composition: Dynamic breakdown of tasks into reusable, composable components.
• Human Oversight: Built-in logic for compliance, safety, and fallback to human review.
• Continual Learning: Automatic evolution of task graphs based on successful or failed trajectories.

A RAGWorker is a core building block used to implement Retrieval Augmented Generation (RAG). It enables a bot to retrieve relevant information from internal documentation (like policies, FAQs, or product info) and compose an answer based on that context.

The worker operates as a pipeline consisting of three stages:

1. Start Node: The entry point of the workflow.
2. Retriever Node: Uses RetrieveEngine.retrieve to find relevant information by applying FAISS (Facebook AI Similarity Search) on documents located in the path specified by the DATA_DIR environment variable.
3. Tool Generator Node: Uses ToolGenerator.context_generate to construct a response using the retrieved context and the user's question.

The internal logic is managed via a LangGraph StateGraph using MessageState.

graph LR;
    start["START"]--"retrieve()"-->retriever["retriever"]--"context_generate()"-->tool_generator["tool_generator"];

A TaskGraph is generated from a Config file (a JSON document). The Generator processes the configuration to create a structured sequence of tasks matched to specific workers.

The generation lifecycle follows these steps:

1. Generate high-level tasks: The Generator uses the role, user_objective, domain, intro, and task_docs from the Config to identify the primary tasks the bot must handle.
2. Generate task planning: High-level tasks that are not directly actionable by a provided Worker are broken down into multiple granular steps or instructions.
3. Interactive with builder: The Generator provides an interactive command-line interface (using textual) allowing a human 'builder' to add, delete, or modify tasks and steps.
4. Fine-tune steps: The generated task planning is refined based on the builder_objective (which may contain specific strategies or hidden objectives).
5. Match tasks with workers: The final plan is mapped to the workers defined in the Config. If no workers are specified, steps are assigned to default workers.

The generator method follows a specific logic flow to produce a response:

1. Unpack MessageState: It extracts user_message, orchestrator_message, and existing message_flow (concatenating response and message_flow).
2. Check for Direct Response: It inspects orchestrator_message.attribute. If direct_response is set to True, it returns the orchestrator_message.message immediately without LLM generation.
3. Determine Prompt Strategy:
   • If message_flow contains existing text, it uses a message_flow_generator_prompt to build upon the existing flow.
   • If no flow exists, it uses a standard message_generator_prompt to generate a response from scratch.
4. LLM Invocation: The prompt is chunked based on the model's context limits, passed through an LLM chain (self.llm | StrOutputParser()), and the resulting answer is saved to state["response"] while clearing state["message_flow"].

def generator(self, state: MessageState) -> MessageState:
    # get the input message
    user_message = state['user_message']
    orchestrator_message = state['orchestrator_message']
    message_flow = state.get('response', "") + "\n" + state.get("message_flow", "")

    # get the orchestrator message content
    orch_msg_content = orchestrator_message.message
    orch_msg_attr = orchestrator_message.attribute
    direct_response = orch_msg_attr.get('direct_response', False)

    if direct_response:
        return orch_msg_content
    
    if message_flow and message_flow != "\n":
        prompt = PromptTemplate.from_template(message_flow_generator_prompt)
        input_prompt = prompt.invoke({
            "sys_instruct": state["sys_instruct"], 
            "message": orch_msg_content, 
            "formatted_chat": user_message.history, 
            "initial_response": message_flow
        })
    else:
        prompt = PromptTemplate.from_template(message_generator_prompt)
        input_prompt = prompt.invoke({
            "sys_instruct": state["sys_instruct"], 
            "message": orch_msg_content, 
            "formatted_chat": user_message.history
        })

    logger.info(f"Prompt: {input_prompt.text}")
    chunked_prompt = chunk_string(input_prompt.text, tokenizer=MODEL["tokenizer"], max_length=MODEL["context"])
    final_chain = self.llm | StrOutputParser()
    answer = final_chain.invoke(chunked_prompt)

    state["message_flow"] = ""
    state["response"] = answer
    return state