Spaces:
Build error
Build error
| import os | |
| import sys | |
| import tempfile | |
| import requests | |
| from basic_agent import BasicAgent, DEFAULT_API_URL | |
| from langchain_core.messages import HumanMessage | |
| from langfuse.langchain import CallbackHandler | |
| # Initialize Langfuse CallbackHandler for LangGraph/Langchain (tracing) | |
| try: | |
| langfuse_handler = CallbackHandler() | |
| except Exception as e: | |
| print(f"Warning: Could not initialize Langfuse handler: {e}") | |
| langfuse_handler = None | |
| # Default Task ID (replace with your desired one or pass via CLI) | |
| DEFAULT_TASK_ID = "f918266a-b3e0-4914-865d-4faa564f1aef" | |
| def fetch_question_by_id(task_id: str, api_base: str = DEFAULT_API_URL): | |
| """Return JSON of a question for a given task_id. | |
| The scoring API does not (yet) expose an explicit /question/{id} endpoint, | |
| so we fetch the full /questions list and filter locally. This works fine | |
| because the list is small (<100 items). | |
| """ | |
| try: | |
| resp = requests.get(f"{api_base}/questions", timeout=30) | |
| resp.raise_for_status() | |
| questions = resp.json() | |
| except Exception as e: | |
| raise RuntimeError(f"Failed to fetch questions list: {e}") from e | |
| for q in questions: | |
| if str(q.get("task_id")) == str(task_id): | |
| return q | |
| raise ValueError(f"Task ID {task_id} not found in /questions list.") | |
| def maybe_download_file(task_id: str, api_base: str = DEFAULT_API_URL) -> str | None: | |
| """Try to download the file associated with a given task id. Returns local path or None.""" | |
| url = f"{api_base}/files/{task_id}" | |
| try: | |
| resp = requests.get(url, timeout=60) | |
| if resp.status_code != 200: | |
| print(f"No file associated with task {task_id} (status {resp.status_code}).") | |
| return None | |
| # Create temp file with same name from headers if available | |
| filename = resp.headers.get("content-disposition", "").split("filename=")[-1].strip("\"") or f"{task_id}_attachment" | |
| tmp_path = os.path.join(tempfile.gettempdir(), filename) | |
| with open(tmp_path, "wb") as f: | |
| f.write(resp.content) | |
| print(f"Downloaded attachment to {tmp_path}") | |
| return tmp_path | |
| except requests.HTTPError as e: | |
| print(f"Could not download file for task {task_id}: {e}") | |
| except Exception as e: | |
| print(f"Error downloading file: {e}") | |
| return None | |
| def main(): | |
| # Determine the task ID (CLI arg > env var > default) | |
| task_id = ( | |
| sys.argv[1] if len(sys.argv) > 1 else os.environ.get("TASK_ID", DEFAULT_TASK_ID) | |
| ) | |
| print(f"Using task ID: {task_id}") | |
| q = fetch_question_by_id(task_id) | |
| question_text = q["question"] | |
| print("\n=== Specific Question ===") | |
| print(f"Task ID : {task_id}") | |
| print(f"Question: {question_text}") | |
| # Attempt to get attachment if any | |
| maybe_download_file(task_id) | |
| # Run the agent | |
| agent = BasicAgent() | |
| result = agent.agent.invoke({"messages": [HumanMessage(content=question_text)]}, config={"callbacks": [langfuse_handler]}) | |
| if isinstance(result, dict) and "messages" in result and result["messages"]: | |
| answer = result["messages"][-1].content.strip() | |
| else: | |
| answer = str(result) | |
| print("\n=== Agent Answer ===") | |
| print(answer) | |
| if __name__ == "__main__": | |
| main() |