Class CodeWriting

KVCOMM/agents/code_writing.py:13–245

Programming agent that validates peers with internal tests.

@AgentRegistry.register('CodeWriting')
class CodeWriting(Node):
    """Programming agent that validates peers with internal tests."""
    def __init__(
        self,
        id: str | None = None,
        role: str = None,
        domain: str = "",
        llm_name: str = "",
        llm_config: KVCommConfig | None = None,
    ):
        super().__init__(id, "CodeWriting" ,domain, llm_name)
        prefix = ""
        self.llm = LLMRegistry.get(llm_name, prefix=prefix, llm_config=llm_config)
        self.prompt_set = PromptSetRegistry.get(domain)
        self.role = self.prompt_set.get_role() if role is None else role
        self.constraint = self.prompt_set.get_constraint(self.role) 
        self.llm.set_id(self.id, self.role)

    async def _process_inputs(
        self,
        raw_inputs:Dict[str,str],
        spatial_info:Dict[str,Dict],
        temporal_info:Dict[str,Dict],
        mode: str = "default",
        **kwargs,
    )->Dict[str, Any]:
        """Prepare prompts, run quick internal checks, and return mode hints."""
        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"
            shared_memory = self.llm._ensure_agent_memory(self.id)

            if self.llm.has_prefix_initialized(self.id) and "placeholder_info" in shared_memory:
                for agent_id, info in spatial_info.items():
                    if (
                        self.role != 'Normal Programmer'
                        and self.role != 'Stupid Programmer'
                        and info['role'] != 'Algorithm Designer'
                    ):
                        agent_mem = self.llm._ensure_agent_memory(agent_id)
                        if raw_inputs['task'] in agent_mem.get('condition', {}):
                            continue
                        code = info['output'].split("```python\n")[-1].split("\n```")[0]
                        is_solved, feedback, _ = PyExecutor().execute(code, self.internal_tests, timeout=1)
                        condition_text = (
                            "Whether it passes internal testing?\n"
                            f"{is_solved}.\n\nThe feedback is:\n\n {feedback}."
                        )
                        self.llm.update_condition_anchor(
                            request_uid=request_uid,
                            owner_agent_id=agent_id,
                            message=raw_inputs['task'],
                            content=condition_text,
                            prefix_text="Whether it passes internal testing?\n",
                        )