RomanNum3ral
/
honcho
mirror of https://github.com/plastic-labs/honcho.git
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

chore: tuning beam bench

This commit is contained in:
Benjamin McCormick 2025-11-20 17:19:14 -05:00
parent 1616304b40
commit f333e64f12
2 changed files with 158 additions and 78 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

170
tests/bench/beam.py
View File

@ -53,7 +53,7 @@ Optional arguments:
--pool-size: Number of Honcho instances in the pool (default: 1)
--batch-size: Number of conversations to run concurrently in each batch (default: 1)
--json-output: Path to write JSON summary results for analytics
--cleanup-workspace: Delete workspace after executing each conversation (default: False)
--cleanup-workspace: Delete workspace after executing each conversation (default: True)
--use-get-context: Use get_context + judge LLM instead of dialectic .chat endpoint (default: False)
```
@ -575,6 +575,86 @@ Extract the ordered list of events or items mentioned in the response. Preserve
alignment.append(best_match_idx)
return alignment
async def _process_single_question(
self,
session: Any,
user_peer: Any,
ability: str,
q_idx: int,
q_data: dict[str, Any],
semaphore: asyncio.Semaphore,
) -> QuestionResult:
"""Process a single BEAM question."""
async with semaphore:
question = q_data["question"]
rubric = q_data.get("rubric", [])
answer = (
q_data.get("answer")
or q_data.get("ideal_response")
or q_data.get("ideal_answer")
)
print(f" [{ability}] Q{q_idx + 1}: {question[:100]}...")
# Execute question using dialectic
if self.use_get_context:
context = await session.get_context(
summary=True,
peer_target="user",
last_user_message=question,
)
context_messages = context.to_anthropic(assistant="assistant")
context_messages.append({"role": "user", "content": question})
response = await self.anthropic_client.messages.create(
model="claude-sonnet-4-5",
max_tokens=2048,
messages=cast(list[MessageParam], context_messages),
)
if not response.content:
actual_response = ""
else:
content_block = response.content[0]
actual_response = getattr(content_block, "text", "")
else:
actual_response = await user_peer.chat(question)
actual_response = (
actual_response if isinstance(actual_response, str) else ""
)
# Judge response based on memory ability
if ability == "event_ordering":
judgment = await self.judge_event_ordering(
question, rubric, actual_response
)
nugget_scores = None
else:
judgment = await self.judge_nugget_based(
question, rubric, actual_response, ability
)
nugget_scores = judgment.get("nugget_scores")
score = judgment.get("overall_score", 0.0)
reasoning = judgment.get("overall_reasoning", "")
question_result: QuestionResult = {
"question": question,
"answer": answer,
"actual_response": actual_response,
"memory_ability": ability,
"rubric": rubric,
"nugget_scores": nugget_scores,
"score": score,
"passed": score >= 0.5,
"reasoning": reasoning,
}
status = "PASS" if score >= 0.5 else "FAIL"
print(f" [{ability}] Q{q_idx + 1} Score: {score:.2f} [{status}]")
return question_result
async def execute_conversation(
self, context_length: str, conversation_id: str, honcho_url: str
) -> ConversationResult:
@ -739,82 +819,28 @@ Extract the ordered list of events or items mentioned in the response. Preserve
print(f"[{workspace_id}] Deriver queue empty. Executing questions...")
# Execute questions for each memory ability
question_tasks: list[Any] = []
semaphore = asyncio.Semaphore(5)
for ability, questions in questions_data.items():
print(
f"\n[{workspace_id}] Testing {ability} ({len(questions)} questions)"
f"\n[{workspace_id}] Queuing {ability} ({len(questions)} questions)"
)
for q_idx, q_data in enumerate(questions):
question = q_data["question"]
rubric = q_data.get("rubric", [])
answer = (
q_data.get("answer")
or q_data.get("ideal_response")
or q_data.get("ideal_answer")
question_tasks.append(
self._process_single_question(
session,
user_peer,
ability,
q_idx,
q_data,
semaphore,
)
)
print(f" Q{q_idx + 1}: {question[:100]}...")
# Execute question using dialectic
if self.use_get_context:
context = await session.get_context(
summary=True,
peer_target="user",
last_user_message=question,
)
context_messages = context.to_anthropic(assistant="assistant")
context_messages.append({"role": "user", "content": question})
from typing import cast
response = await self.anthropic_client.messages.create(
model="claude-sonnet-4-5",
max_tokens=2048,
messages=cast(list[MessageParam], context_messages),
)
if not response.content:
actual_response = ""
else:
content_block = response.content[0]
actual_response = getattr(content_block, "text", "")
else:
actual_response = await user_peer.chat(question)
actual_response = (
actual_response if isinstance(actual_response, str) else ""
)
# Judge response based on memory ability
if ability == "event_ordering":
judgment = await self.judge_event_ordering(
question, rubric, actual_response
)
nugget_scores = None
else:
judgment = await self.judge_nugget_based(
question, rubric, actual_response, ability
)
nugget_scores = judgment.get("nugget_scores")
score = judgment.get("overall_score", 0.0)
reasoning = judgment.get("overall_reasoning", "")
question_result: QuestionResult = {
"question": question,
"answer": answer,
"actual_response": actual_response,
"memory_ability": ability,
"rubric": rubric,
"nugget_scores": nugget_scores,
"score": score,
"passed": score >= 0.5,
"reasoning": reasoning,
}
result["question_results"].append(question_result)
status = "PASS" if score >= 0.5 else "FAIL"
print(f" Score: {score:.2f} [{status}]")
results = await asyncio.gather(*question_tasks)
result["question_results"] = list(results)
# Calculate ability scores
ability_totals: dict[str, list[float]] = {}
@ -1064,7 +1090,7 @@ async def main() -> int:
parser.add_argument(
"--batch-size",
type=int,
default=1,
default=10,
help="Number of conversations to run concurrently in each batch (default: 1)",
)
@ -1077,7 +1103,7 @@ async def main() -> int:
parser.add_argument(
"--cleanup-workspace",
action="store_true",
help="Delete workspace after executing each conversation (default: False)",
help="Delete workspace after executing each conversation (default: True)",
)
parser.add_argument(

66
tests/bench/harness.py
View File

@ -68,19 +68,55 @@ class HonchoHarness:
# Update the database port
compose_data["services"]["database"]["ports"] = [f"{self.db_port}:5432"]
# Increase shared memory size
compose_data["services"]["database"]["shm_size"] = "4gb"
# Configure Postgres for larger workloads
cmd = ["postgres"]
params = {
"max_connections": "800",
"max_wal_size": "8GB",
"min_wal_size": "2GB",
"checkpoint_timeout": "15min",
"maintenance_work_mem": "1GB",
"work_mem": "64MB",
}
for k, v in params.items():
cmd.extend(["-c", f"{k}={v}"])
compose_data["services"]["database"]["command"] = cmd
# Update the Redis port
# TODO: Make this configurable if running multiple instances
compose_data["services"]["redis"]["ports"] = ["6379:6379"]
# Add a unique project name to avoid conflicts
compose_data["name"] = f"honcho_harness_{self.db_port}"
# remove init.sql mount since we use provision_db.py for setup
# Configure named volumes for better performance and cleanup
compose_data.setdefault("volumes", {})
db_vol_name = f"pgdata_{self.db_port}"
redis_vol_name = f"redis_data_{self.db_port}"
compose_data["volumes"][db_vol_name] = {}
compose_data["volumes"][redis_vol_name] = {}
# Update database volumes - use named volume instead of bind mount
new_db_volumes: list[str] = []
if "volumes" in compose_data["services"]["database"]:
compose_data["services"]["database"]["volumes"] = [
vol
for vol in compose_data["services"]["database"]["volumes"]
if "init.sql" not in vol
]
for vol in compose_data["services"]["database"]["volumes"]:
# Filter out init.sql (handled by provision script) and existing data mounts
if "init.sql" not in vol and ":/var/lib/postgresql/data/" not in vol:
new_db_volumes.append(vol)
new_db_volumes.append(f"{db_vol_name}:/var/lib/postgresql/data/")
compose_data["services"]["database"]["volumes"] = new_db_volumes
# Update redis volumes - use named volume
new_redis_volumes: list[str] = []
if "volumes" in compose_data["services"]["redis"]:
for vol in compose_data["services"]["redis"]["volumes"]:
if ":/data" not in vol:
new_redis_volumes.append(vol)
new_redis_volumes.append(f"{redis_vol_name}:/data")
compose_data["services"]["redis"]["volumes"] = new_redis_volumes
# Create temporary file
self.temp_dir = Path(tempfile.mkdtemp(prefix="honcho_harness_"))
@ -134,6 +170,24 @@ class HonchoHarness:
"""
print(f"Starting PostgreSQL database on port {self.db_port}...")
# Ensure clean state by removing any existing containers/volumes
# This handles cases where a previous run was interrupted
subprocess.run(
[
"docker",
"compose",
"-f",
str(self.docker_compose_file),
"-p",
f"honcho_harness_{self.db_port}",
"down",
"--volumes",
"--remove-orphans",
],
cwd=self.temp_dir,
capture_output=True,
)
# Change to the temp directory and start the database service
result = subprocess.run(
[