Summarize peer outputs, running tests on code blocks when present.
(
self,
raw_inputs: Dict[str, str],
spatial_info: Dict[str, Any],
internal_tests: List[str],
)
| 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], |
| 71 | ) -> str: |
| 72 | """Summarize peer outputs, running tests on code blocks when present.""" |
| 73 | paragraphs: List[str] = [] |
| 74 | for agent_id, info in spatial_info.items(): |
| 75 | role = info.get("role", "agent") |
| 76 | output = info.get("output") |
| 77 | if not isinstance(output, str): |
| 78 | paragraphs.append(f"Agent {agent_id} as a {role} returned invalid output.\n\n") |
| 79 | continue |
| 80 | if self._is_python_code_block(output): |
| 81 | code = self._extract_python_code(output) |
| 82 | is_solved, feedback, _ = self._executor.execute(code, internal_tests, timeout=10) |
| 83 | paragraphs.append( |
| 84 | f"Agent {agent_id} as a {role}:\n\n" |
| 85 | f"The code written by the agent is:\n\n{output}\n\n" |
| 86 | f"Whether it passes internal testing? {is_solved}.\n\n" |
| 87 | f"The feedback is:\n\n{feedback}.\n\n" |
| 88 | ) |
| 89 | continue |
| 90 | paragraphs.append( |
| 91 | f"Agent {agent_id} as a {role} provides the following info: {output}\n\n" |
| 92 | ) |
| 93 | return "".join(paragraphs) |
| 94 | |
| 95 | async def _process_inputs( |
| 96 | self, |
nothing calls this directly
no test coverage detected