Final code synthesis agent that integrates peers and executes tests.
| 11 | |
| 12 | @AgentRegistry.register('FinalWriteCode') |
| 13 | class FinalWriteCode(Node): |
| 14 | """Final code synthesis agent that integrates peers and executes tests.""" |
| 15 | def __init__( |
| 16 | self, |
| 17 | id: str | None = None, |
| 18 | domain: str = "", |
| 19 | llm_name: str = "", |
| 20 | llm_config: KVCommConfig | None = None, |
| 21 | ): |
| 22 | super().__init__(id, "FinalWriteCode" ,domain, llm_name) |
| 23 | prefix = "" |
| 24 | self.llm = LLMRegistry.get(llm_name, prefix=prefix, llm_config=llm_config) |
| 25 | self.role = 'FinalWriteCode' |
| 26 | self.llm.set_id(self.id, 'FinalWriteCode') |
| 27 | self.prompt_set = PromptSetRegistry.get(domain) |
| 28 | self._executor = PyExecutor() |
| 29 | |
| 30 | @staticmethod |
| 31 | def extract_example(prompt: Dict[str, Any] | str) -> List[str]: |
| 32 | """Return doctest-style assertions extracted from the task description.""" |
| 33 | if isinstance(prompt, dict): |
| 34 | prompt_text = str(prompt.get("task", "")) |
| 35 | else: |
| 36 | prompt_text = str(prompt) |
| 37 | |
| 38 | lines = [line.strip() for line in prompt_text.splitlines() if line.strip()] |
| 39 | results: List[str] = [] |
| 40 | iterator = iter(lines) |
| 41 | for line in iterator: |
| 42 | if not line.startswith(">>>"): |
| 43 | continue |
| 44 | function_call = line[4:].strip() |
| 45 | expected_output = next(iterator, "").strip() |
| 46 | if not function_call or not expected_output: |
| 47 | continue |
| 48 | results.append(f"assert {function_call} == {expected_output}") |
| 49 | return results |
| 50 | |
| 51 | @staticmethod |
| 52 | def _is_python_code_block(text: str) -> bool: |
| 53 | text = text.strip() |
| 54 | return text.startswith("```python") and text.endswith("```") |
| 55 | |
| 56 | @staticmethod |
| 57 | def _extract_python_code(text: str) -> str: |
| 58 | """Extract pure python code from a fenced block.""" |
| 59 | if not FinalWriteCode._is_python_code_block(text): |
| 60 | return text |
| 61 | content = text.strip()[len("```python") :].strip() |
| 62 | if content.endswith("```"): |
| 63 | content = content[:-3].strip() |
| 64 | return content |
| 65 | |
| 66 | def _summarize_agent_outputs( |
| 67 | self, |
| 68 | raw_inputs: Dict[str, str], |
| 69 | spatial_info: Dict[str, Any], |
| 70 | internal_tests: List[str], |
nothing calls this directly
no outgoing calls
no test coverage detected