Class FinalRefer

KVCOMM/agents/final_decision.py:337–608

Final referencing/answer agent assembling the final response.

@AgentRegistry.register('FinalRefer')
class FinalRefer(Node):
    """Final referencing/answer agent assembling the final response."""
    def __init__(
        self,
        id: str | None = None,
        domain: str = "",
        llm_name: str = "",
        llm_config: KVCommConfig | None = None,
    ):
        super().__init__(id, "FinalRefer" ,domain, llm_name)
        prefix = ""
        self.llm = LLMRegistry.get(llm_name, prefix=prefix, llm_config=llm_config)
        self.role = 'FinalRefer'
        self.llm.set_id(self.id, 'FinalRefer')
        self.prompt_set = PromptSetRegistry.get(domain)

    async def _process_inputs(
        self,
        raw_inputs:Dict[str,str],
        spatial_info:Dict[str,Any],
        temporal_info:Dict[str,Any],
        mode: str = "default",
        **kwargs,
    )->Dict[str, Any]:
        """ To be overriden by the descendant class """
        """ Process the raw_inputs(most of the time is a List[Dict]) """
        if mode == "allow_kv_reuse":
            request_uid = raw_inputs.get("_request_uid") or kwargs.get("request_uid")
            if request_uid is None:
                raise ValueError("request_uid is required for request-scoped anchor updates.")

            preferred_mode = "kv_reuse"
            agent_memory = self.llm._ensure_agent_memory(self.id)
            prefix_text = kwargs.get("prefix", "")

            has_shared_prefix = (
                self.llm.has_prefix_initialized(self.id)
                and "placeholder_info" in agent_memory
            )
            early_response: str | None = None

            if has_shared_prefix:
                task = raw_inputs["task"]
                for agent_id, info in spatial_info.items():
                    cond_text: str | None = None
                    cond_prefix: str | None = None

                    if self.domain == "gsm8k" and info["role"] == "Programming Expert":
                        answer = execute_code_get_return(
                            info["output"].split("```python\n")[-1].split("\n```")[0]
                        )
                        if answer is None:
                            answer = "No variable is named answer."
                        cond_text = f"the answer is {answer}"
                        cond_prefix = "the answer is "
                    elif (
                        self.domain == "humaneval"
                        and self.role not in {"Normal Programmer", "Stupid Programmer"}