build-small-hackathon/hackathon-advisor
16
1from __future__ import annotations2 3from dataclasses import dataclass4import json5from typing import Any, Literal6from xml.etree import ElementTree7 8 9JsonType = Literal["string", "integer", "number", "boolean", "array", "object"]10 11 12@dataclass(frozen=True)13class ToolField:14 type: JsonType15 description: str16 required: bool = False17 enum: tuple[str, ...] = ()18 items_type: JsonType | None = None19 20 def to_schema(self) -> dict[str, Any]:21 schema: dict[str, Any] = {22 "type": self.type,23 "description": self.description,24 }25 if self.enum:26 schema["enum"] = list(self.enum)27 if self.items_type:28 schema["items"] = {"type": self.items_type}29 return schema30 31 32@dataclass(frozen=True)33class ToolSpec:34 name: str35 description: str36 fields: dict[str, ToolField]37 38 def to_schema(self) -> dict[str, Any]:39 return {40 "type": "function",41 "function": {42 "name": self.name,43 "description": self.description,44 "parameters": {45 "type": "object",46 "additionalProperties": False,47 "properties": {48 name: field.to_schema() for name, field in self.fields.items()49 },50 "required": [name for name, field in self.fields.items() if field.required],51 },52 },53 }54 55 56@dataclass(frozen=True)57class ToolCall:58 name: str59 arguments: dict[str, Any]60 61 def to_dict(self) -> dict[str, Any]:62 return {"name": self.name, "arguments": self.arguments}63 64 65@dataclass(frozen=True)66class ToolResolution:67 status: Literal["valid", "defaulted"]68 call: ToolCall69 errors: tuple[str, ...]70 71 def to_dict(self) -> dict[str, Any]:72 return {73 "status": self.status,74 "call": self.call.to_dict(),75 "errors": list(self.errors),76 }77 78 79class ToolContractError(ValueError):80 pass81 82 83TOOL_SPECS: dict[str, ToolSpec] = {84 "list_projects": ToolSpec(85 name="list_projects",86 description="Read prominent project cards from the offline snapshot.",87 fields={88 "track": ToolField("string", "Optional model, goal, or topic filter."),89 "sort": ToolField("string", "Sort key.", enum=("likes", "recent", "title")),90 },91 ),92 "search_projects": ToolSpec(93 name="search_projects",94 description="Find existing Spaces that echo the user's project idea.",95 fields={"query": ToolField("string", "The user idea or topic to search.", required=True)},96 ),97 "get_project": ToolSpec(98 name="get_project",99 description="Read one project card by full Space id or slug.",100 fields={"id": ToolField("string", "Project id or slug, such as org-name/space-name.", required=True)},101 ),102 "find_whitespace": ToolSpec(103 name="find_whitespace",104 description="Return under-explored project regions from the offline index.",105 fields={},106 ),107 "save_idea": ToolSpec(108 name="save_idea",109 description="Write or update the current idea page.",110 fields={111 "title": ToolField("string", "Short idea title.", required=True),112 "pitch": ToolField("string", "One-sentence idea pitch.", required=True),113 },114 ),115 "score_idea": ToolSpec(116 name="score_idea",117 description="Score the current idea against the fixed originality and build-fit rubric.",118 fields={"id": ToolField("string", "Idea id; omit to score the current idea.")},119 ),120 "compare_ideas": ToolSpec(121 name="compare_ideas",122 description="Rank the current idea board and explain tradeoffs.",123 fields={},124 ),125 "make_plan": ToolSpec(126 name="make_plan",127 description="Draft the next build steps for the current idea.",128 fields={"id": ToolField("string", "Idea id; omit to plan the current idea.")},129 ),130 "update_profile": ToolSpec(131 name="update_profile",132 description="Remember a user skill, constraint, preference, or available time.",133 fields={134 "field": ToolField(135 "string",136 "Profile field to update.",137 required=True,138 enum=("skills", "time", "preferences", "constraints"),139 ),140 "value": ToolField("string", "Profile value to remember.", required=True),141 },142 ),143 "set_goals": ToolSpec(144 name="set_goals",145 description="Change the selected goals used to bias ideation and planning.",146 fields={"goals": ToolField("array", "Goal ids to prioritize.", required=True, items_type="string")},147 ),148}149 150 151def tool_schemas() -> list[dict[str, Any]]:152 return [spec.to_schema() for spec in TOOL_SPECS.values()]153 154 155def parse_xml_tool_call(text: str) -> ToolCall:156 wrapped = f"<root>{text.strip()}</root>"157 try:158 root = ElementTree.fromstring(wrapped)159 except ElementTree.ParseError as error:160 raise ToolContractError(f"invalid XML tool call: {error}") from error161 162 functions = [node for node in root if node.tag == "function"]163 if len(functions) != 1:164 raise ToolContractError(f"expected exactly one function call, got {len(functions)}")165 node = functions[0]166 name = str(node.attrib.get("name") or "").strip()167 if not name:168 raise ToolContractError("function call is missing a name")169 raw_arguments = (node.text or "").strip() or "{}"170 try:171 arguments = json.loads(raw_arguments)172 except json.JSONDecodeError as error:173 raise ToolContractError(f"function arguments are not valid JSON: {error.msg}") from error174 if not isinstance(arguments, dict):175 raise ToolContractError("function arguments must be a JSON object")176 return ToolCall(name=name, arguments=arguments)177 178 179def validate_tool_call(call: ToolCall, specs: dict[str, ToolSpec] = TOOL_SPECS) -> ToolCall:180 spec = specs.get(call.name)181 if spec is None:182 raise ToolContractError(f"unknown tool: {call.name}")183 allowed = set(spec.fields)184 extra = sorted(set(call.arguments) - allowed)185 if extra:186 raise ToolContractError(f"unexpected arguments for {call.name}: {', '.join(extra)}")187 missing = sorted(name for name, field in spec.fields.items() if field.required and name not in call.arguments)188 if missing:189 raise ToolContractError(f"missing required arguments for {call.name}: {', '.join(missing)}")190 for name, value in call.arguments.items():191 field = spec.fields[name]192 _validate_value(call.name, name, value, field)193 return call194 195 196def resolve_tool_call(model_output: str, fallback_query: str = "") -> ToolResolution:197 errors: list[str] = []198 try:199 call = validate_tool_call(parse_xml_tool_call(model_output))200 return ToolResolution(status="valid", call=call, errors=())201 except ToolContractError as error:202 errors.append(str(error))203 204 query = fallback_query.strip()205 if query:206 call = ToolCall("search_projects", {"query": query})207 else:208 call = ToolCall("find_whitespace", {})209 return ToolResolution(status="defaulted", call=call, errors=tuple(errors))210 211 212def _validate_value(tool_name: str, field_name: str, value: Any, field: ToolField) -> None:213 if field.type == "string":214 valid = isinstance(value, str)215 elif field.type == "integer":216 valid = isinstance(value, int) and not isinstance(value, bool)217 elif field.type == "number":218 valid = (isinstance(value, int | float)) and not isinstance(value, bool)219 elif field.type == "boolean":220 valid = isinstance(value, bool)221 elif field.type == "array":222 valid = isinstance(value, list)223 elif field.type == "object":224 valid = isinstance(value, dict)225 else:226 valid = False227 if not valid:228 raise ToolContractError(f"{tool_name}.{field_name} must be {field.type}")229 if field.enum and value not in field.enum:230 raise ToolContractError(f"{tool_name}.{field_name} must be one of: {', '.join(field.enum)}")231 if field.items_type and isinstance(value, list):232 for index, item in enumerate(value):233 _validate_value(tool_name, f"{field_name}[{index}]", item, ToolField(field.items_type, "array item"))234 