Spaces:
Build error
Build error
| import os | |
| 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 | |
| def fetch_random_question(api_base: str = DEFAULT_API_URL): | |
| """Return JSON of /random-question.""" | |
| resp = requests.get(f"{api_base}/random-question", timeout=30) | |
| resp.raise_for_status() | |
| return resp.json() | |
| 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(): | |
| q = fetch_random_question() | |
| task_id = str(q["task_id"]) | |
| question_text = q["question"] | |
| print("\n=== Random 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() |