File size: 4,990 Bytes
de36225 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 | """
Vision MCP tool - visual Q&A via Groq Vision API (llama-3.2-11b-vision-preview).
Cloud path : image -> Groq Llama 3.2 Vision -> structured answer (FREE, 14400 req/day)
Local path : image -> Ollama qwen2-vl -> answer (dev only)
Registered in mcp_server.py as 'granite_vision.qa'.
"""
import os
import base64
import requests
from typing import Optional
def qa(image_path: str, question: str, model_path: Optional[str] = None) -> dict:
"""Answer a question about a document image using Groq Vision or local Ollama."""
print(f"[vision.qa] image={image_path} question={question[:80]}")
# CLOUD MODE: Groq Vision API
if os.environ.get("USE_HF_CLOUD", "0") in ("1", "true", "True"):
groq_key = os.environ.get("GROQ_API_KEY")
groq_model = os.environ.get("GROQ_VISION_MODEL", "meta-llama/llama-4-scout-17b-16e-instruct")
if groq_key:
print(f"[granite_vision.qa] CLOUD MODE: {groq_model} via Groq")
try:
from groq import Groq
client = Groq(api_key=groq_key)
with open(image_path, "rb") as f:
img_b64 = base64.b64encode(f.read()).decode("utf-8")
ext = os.path.splitext(image_path)[-1].lower().lstrip(".")
mime = "image/jpeg" if ext in ("jpg", "jpeg") else f"image/{ext}"
response = client.chat.completions.create(
model=groq_model,
messages=[
{
"role": "user",
"content": [
{
"type": "text",
"text": (
"You are analyzing a document image. "
"Look carefully at the layout, tables, numbers, and text. "
"Answer the following question based only on what you see:\n\n"
+ question
)
},
{
"type": "image_url",
"image_url": {
"url": f"data:{mime};base64,{img_b64}"
}
}
]
}
],
max_tokens=512
)
answer = response.choices[0].message.content.strip()
print(f"[granite_vision.qa] Groq answer ({len(answer)} chars): {answer[:200]}")
return {"answer": answer, "model": f"{groq_model} (Groq Vision)"}
except Exception as e:
print(f"[granite_vision.qa] Groq failed: {e}")
return {
"answer": f"Groq Vision error: {e}",
"model": groq_model,
"error": str(e)
}
else:
print("[granite_vision.qa] GROQ_API_KEY not set in HF secrets.")
return {
"answer": "GROQ_API_KEY is not configured. Please add it in HuggingFace Space secrets.",
"model": "none",
"error": "missing_groq_key"
}
# LOCAL MODE (dev only): Ollama
local_model = os.environ.get("GRANITE_MODEL_PATH", "qwen2-vl:2b")
print(f"[granite_vision.qa] LOCAL MODE: sending image to Ollama ({local_model})")
try:
with open(image_path, "rb") as f:
img_b64 = base64.b64encode(f.read()).decode("utf-8")
base_url = os.environ.get("OLLAMA_BASE_URL", "http://localhost:11434").rstrip("/")
url = f"{base_url}/api/chat"
payload = {
"model": local_model,
"messages": [
{
"role": "user",
"content": f"Carefully analyze this document image. {question}",
"images": [img_b64]
}
],
"stream": False,
"keep_alive": 60,
"options": {"num_ctx": 1024, "temperature": 0.1}
}
r = requests.post(url, json=payload, timeout=300)
if r.status_code == 200:
answer = r.json().get("message", {}).get("content", "").strip()
print(f"[granite_vision.qa] Ollama answer ({len(answer)} chars): {answer[:200]}")
return {"answer": answer, "model": f"{local_model} (Ollama)"}
else:
msg = f"Ollama HTTP {r.status_code}: {r.text}"
print(f"[granite_vision.qa] {msg}")
return {"answer": msg, "model": f"{local_model} (Ollama)", "error": "http_error"}
except Exception as e:
msg = f"[ERROR] Ollama Q&A failed: {e}"
print(f"[granite_vision.qa] {msg}")
return {"answer": msg, "model": f"{local_model} (Ollama)", "error": str(e)} |