Class IntegrationAgent

integuru/agent.py:13–410

from integuru.models.agent_state import AgentState

class IntegrationAgent:
    ACTION_URL_KEY: str = "action_url"
    IN_PROCESS_NODE_KEY: str = "in_process_node"
    TO_BE_PROCESSED_NODES_KEY: str = "to_be_processed_nodes"
    IN_PROCESS_NODE_DYNAMIC_PARTS_KEY: str = "in_process_node_dynamic_parts"
    MASTER_NODE_KEY: str = "master_node"
    INPUT_VARIABLES_KEY: str = "input_variables"

    def __init__(
        self,
        prompt: str,
        har_file_path: str,
        cookie_path: str,
    ):  
        self.prompt: str = prompt
        self.duplicate_part_set: Set[str] = set()
        self.global_master_node: Optional[str] = None
        self.req_to_res_map: Dict[Request, str] = parse_har_file(har_file_path)
        self.url_to_res_req_dict: Dict[str, Dict[str, Any]] = build_url_to_req_res_map(self.req_to_res_map)
        self.har_urls: List[Tuple[str, str, str, str]] = get_har_urls(har_file_path)
        self.cookie_dict: Dict[str, Dict[str, Any]] = parse_cookie_file_to_dict(cookie_path)
        self.curl_to_id_dict: Dict[str, str] = {}
        self.cookie_to_id_dict: Dict[str, str] = {}
        self.dag_manager: DAGManager = DAGManager()

    def end_url_identify_agent(self, state: AgentState) -> AgentState:
        """
        Identify the URL responsible for a specific action
        """
        function_def = {
            "name": "identify_end_url",
            "description": "Identify the URL responsible for a specific action",
            "parameters": {
                "type": "object",
                "properties": {
                    "url": {
                        "type": "string",
                        "description": f"The URL responsible for {self.prompt}"
                    }
                },
                "required": ["url"]
            }
        }

        prompt = f"""
        {self.har_urls}
        Task:
        Given the above list of URLs, request types, and response formats, find the URL responsible for the action below:
        {self.prompt}
        """

        response = llm.get_instance().invoke(
            prompt,
            functions=[function_def],
            function_call={"name": "identify_end_url"}
        )
        
        function_call = response.additional_kwargs['function_call']