a2a-source-edit: write agent.py
This commit is contained in:
a2a-cloud
742
agent.py
742
agent.py
@@ -1,627 +1,189 @@
|
|||||||
"""QuoteJudge Studio v1: deterministic quote comparison with durable per-user state."""
|
"""quote-judge-studio-v1 agent.
|
||||||
|
|
||||||
|
Starter stack:
|
||||||
|
- DeepAgents for tool-calling orchestration
|
||||||
|
- Caller-provided LLM credentials via ctx.llm
|
||||||
|
- A tiny model-call middleware hook you can replace with tracing,
|
||||||
|
routing, rate limits, or policy checks
|
||||||
|
"""
|
||||||
from __future__ import annotations
|
from __future__ import annotations
|
||||||
|
|
||||||
import asyncio
|
|
||||||
import base64
|
|
||||||
import csv
|
|
||||||
import hashlib
|
|
||||||
import io
|
|
||||||
import json
|
import json
|
||||||
import os
|
from pathlib import Path
|
||||||
import re
|
from typing import Any
|
||||||
from datetime import datetime, timezone
|
|
||||||
from decimal import Decimal, ROUND_HALF_UP
|
|
||||||
from typing import Annotated, Any
|
|
||||||
|
|
||||||
from pydantic import BaseModel, Field, field_validator
|
from pydantic import BaseModel
|
||||||
|
|
||||||
import a2a_pack as a2a
|
import a2a_pack as a2a
|
||||||
from a2a_pack import (
|
from a2a_pack import (
|
||||||
A2AAgent,
|
A2AAgent,
|
||||||
AgentDatabase,
|
LLMProvisioning,
|
||||||
AgentDatabaseEnv,
|
{{ auth_type }},
|
||||||
AgentDatabaseMigrations,
|
|
||||||
AgentPlatformResources,
|
|
||||||
PlatformUserAuth,
|
|
||||||
Pricing,
|
Pricing,
|
||||||
Resources,
|
|
||||||
RunContext,
|
RunContext,
|
||||||
State,
|
|
||||||
WorkspaceAccess,
|
WorkspaceAccess,
|
||||||
WorkspaceMode,
|
WorkspaceMode,
|
||||||
)
|
)
|
||||||
from a2a_pack.workspace import FileUpload, UploadedFile
|
from a2a_pack.context import LLMCreds
|
||||||
|
|
||||||
try: # psycopg is installed in the deployed image via requirements.txt.
|
|
||||||
import psycopg
|
|
||||||
from psycopg.rows import dict_row
|
|
||||||
from psycopg.types.json import Jsonb
|
|
||||||
except Exception: # pragma: no cover - lets `a2a card` run before deps install.
|
|
||||||
psycopg = None
|
|
||||||
dict_row = None
|
|
||||||
Jsonb = None
|
|
||||||
|
|
||||||
|
|
||||||
COMPARISON_ID_RE = re.compile(r"^[A-Za-z0-9][A-Za-z0-9_.-]{0,79}$")
|
|
||||||
MAX_QUOTES = 10
|
|
||||||
MAX_BROWSER_DOCUMENTS = 4
|
|
||||||
MAX_UPLOAD_BYTES = 64 * 1024
|
|
||||||
ALLOWED_UPLOAD_MEDIA_TYPES = {
|
|
||||||
"application/json",
|
|
||||||
"text/json",
|
|
||||||
"text/csv",
|
|
||||||
"application/csv",
|
|
||||||
"application/vnd.ms-excel",
|
|
||||||
"text/plain",
|
|
||||||
}
|
|
||||||
AGENT_VERSION = "0.1.4"
|
|
||||||
DB_CONNECT_TIMEOUT_SECONDS = 5
|
|
||||||
DB_STATEMENT_TIMEOUT_MS = 10_000
|
|
||||||
DB_LOCK_TIMEOUT_MS = 5_000
|
|
||||||
|
|
||||||
|
|
||||||
class QuoteJudgeStudioV1Config(BaseModel):
|
class QuoteJudgeStudioV1Config(BaseModel):
|
||||||
"""Runtime config is intentionally empty; the app uses platform auth + DB."""
|
pass
|
||||||
|
|
||||||
|
|
||||||
class QuoteJudgeState(BaseModel):
|
SYSTEM_PROMPT = """\
|
||||||
"""Card-visible durable state marker; data lives in managed Postgres."""
|
You are a compact tool-calling agent.
|
||||||
|
|
||||||
|
Use the text_stats tool when the user asks about text, counts, summaries,
|
||||||
|
or anything where exact length/word numbers would help. Mention tool results
|
||||||
|
briefly instead of dumping raw JSON.
|
||||||
|
"""
|
||||||
|
|
||||||
|
RUNTIME_SKILLS_DIR = "quote-judge-studio-v1/.deepagents/skills/"
|
||||||
|
DEEPAGENTS_RECURSION_LIMIT = 500
|
||||||
|
|
||||||
|
|
||||||
class QuoteInput(BaseModel):
|
class QuoteJudgeStudioV1(A2AAgent[QuoteJudgeStudioV1Config, {{ auth_type }}]):
|
||||||
vendor: str = Field(..., min_length=1, max_length=120)
|
|
||||||
quantity: int = Field(..., ge=1, le=1_000_000)
|
|
||||||
unit_price: float = Field(..., ge=0, le=1_000_000_000)
|
|
||||||
delivery_days: int = Field(..., ge=0, le=3650)
|
|
||||||
warranty_months: int = Field(..., ge=0, le=240)
|
|
||||||
|
|
||||||
@field_validator("vendor")
|
|
||||||
@classmethod
|
|
||||||
def clean_vendor(cls, value: str) -> str:
|
|
||||||
cleaned = " ".join(str(value or "").split())
|
|
||||||
if not cleaned:
|
|
||||||
raise ValueError("vendor is required")
|
|
||||||
return cleaned
|
|
||||||
|
|
||||||
|
|
||||||
class QuoteWeights(BaseModel):
|
|
||||||
price: float = Field(..., ge=0, le=1000)
|
|
||||||
delivery: float = Field(..., ge=0, le=1000)
|
|
||||||
warranty: float = Field(..., ge=0, le=1000)
|
|
||||||
|
|
||||||
@field_validator("warranty")
|
|
||||||
@classmethod
|
|
||||||
def at_least_one_weight(cls, value: float, info: Any) -> float:
|
|
||||||
values = dict(info.data)
|
|
||||||
total = float(values.get("price") or 0) + float(values.get("delivery") or 0) + float(value or 0)
|
|
||||||
if total <= 0:
|
|
||||||
raise ValueError("at least one weight must be greater than zero")
|
|
||||||
return value
|
|
||||||
|
|
||||||
|
|
||||||
class BrowserDocument(BaseModel):
|
|
||||||
filename: str = Field(..., min_length=1, max_length=160)
|
|
||||||
media_type: str = Field(..., min_length=1, max_length=120)
|
|
||||||
data_base64: str = Field(..., min_length=1, max_length=90_000)
|
|
||||||
|
|
||||||
@field_validator("filename")
|
|
||||||
@classmethod
|
|
||||||
def safe_filename(cls, value: str) -> str:
|
|
||||||
cleaned = str(value or "").strip().replace("\\", "/")
|
|
||||||
if not cleaned or "/" in cleaned or cleaned in {".", ".."}:
|
|
||||||
raise ValueError("filename must be a simple file name")
|
|
||||||
return cleaned
|
|
||||||
|
|
||||||
@field_validator("media_type")
|
|
||||||
@classmethod
|
|
||||||
def known_media_type(cls, value: str) -> str:
|
|
||||||
cleaned = str(value or "").split(";", 1)[0].strip().lower()
|
|
||||||
if cleaned not in ALLOWED_UPLOAD_MEDIA_TYPES:
|
|
||||||
raise ValueError("unsupported media type")
|
|
||||||
return cleaned
|
|
||||||
|
|
||||||
|
|
||||||
class QuoteJudgeStudioV1(A2AAgent[QuoteJudgeStudioV1Config, PlatformUserAuth]):
|
|
||||||
name = "quote-judge-studio-v1"
|
name = "quote-judge-studio-v1"
|
||||||
description = (
|
description = "One-page QuoteJudge full-stack startup for comparing vendor quotes with managed persistence."
|
||||||
"QuoteJudge compares vendor quotes with weighted price, delivery, and warranty "
|
version = "0.1.0"
|
||||||
"scoring, persists per-user comparisons, supports browser uploads, and serves "
|
|
||||||
"a polished packed product UI."
|
|
||||||
)
|
|
||||||
version = AGENT_VERSION
|
|
||||||
|
|
||||||
config_model = QuoteJudgeStudioV1Config
|
config_model = QuoteJudgeStudioV1Config
|
||||||
auth_model = PlatformUserAuth
|
auth_model = {{ auth_type }}
|
||||||
|
|
||||||
state = State.DURABLE
|
# Hosted generated agents read the caller's saved LLM credential through
|
||||||
state_model = QuoteJudgeState
|
# ctx.llm. The platform may proxy that credential through LiteLLM, but agent
|
||||||
|
# code never reads provider keys, LiteLLM master keys, or OPENAI_API_KEY
|
||||||
|
# directly.
|
||||||
|
llm_provisioning = LLMProvisioning.PLATFORM
|
||||||
pricing = Pricing(
|
pricing = Pricing(
|
||||||
price_per_call_usd=0.0,
|
price_per_call_usd=0.0,
|
||||||
caller_pays_llm=False,
|
caller_pays_llm=True,
|
||||||
notes="Deterministic quote scoring; no LLM credential or provider key is required.",
|
notes="Starter agent uses the caller's saved LLM credential via ctx.llm.",
|
||||||
)
|
)
|
||||||
resources = Resources(cpu="500m", memory="512Mi", max_runtime_seconds=600)
|
|
||||||
tools_used = ("postgres", "mcp", "packed-react-frontend")
|
|
||||||
workspace_access = WorkspaceAccess.dynamic(
|
workspace_access = WorkspaceAccess.dynamic(
|
||||||
max_files=8,
|
max_files=64,
|
||||||
allowed_modes=(WorkspaceMode.READ_ONLY,),
|
allowed_modes=(WorkspaceMode.READ_ONLY, WorkspaceMode.READ_WRITE_OVERLAY),
|
||||||
require_reason=False,
|
require_reason=False,
|
||||||
max_total_size_bytes=MAX_UPLOAD_BYTES * MAX_BROWSER_DOCUMENTS,
|
|
||||||
)
|
|
||||||
platform_resources = AgentPlatformResources(
|
|
||||||
databases=(
|
|
||||||
AgentDatabase(
|
|
||||||
name="quote-judge-studio-v1-data",
|
|
||||||
scope="user",
|
|
||||||
access_mode="read_write",
|
|
||||||
env=AgentDatabaseEnv(url="DATABASE_URL"),
|
|
||||||
migrations=AgentDatabaseMigrations(path="db/migrations"),
|
|
||||||
),
|
|
||||||
)
|
|
||||||
)
|
)
|
||||||
|
tools_used = ("deepagents", "langchain")
|
||||||
|
|
||||||
@a2a.tool(
|
@a2a.tool(description="Ask the starter DeepAgent to answer with tool calls when useful")
|
||||||
description="Compare two to ten vendor quotes with weighted price, delivery, and warranty scoring; persist the result and a production receipt.",
|
async def ask(self, ctx: RunContext[{{ auth_type }}], prompt: str) -> str:
|
||||||
timeout_seconds=30,
|
creds = ctx.llm
|
||||||
idempotent=True,
|
await ctx.emit_progress(f"llm: {creds.model} via {creds.source}")
|
||||||
cost_class="deterministic",
|
if not creds.api_key:
|
||||||
|
return (
|
||||||
|
"LLM key required. Add an LLM credential in Settings > LLM "
|
||||||
|
"credentials before running this agent; for local --invoke "
|
||||||
|
"runs set AGENT_LLM_KEY."
|
||||||
)
|
)
|
||||||
async def compare_quotes(
|
graph = self._build_deep_agent(ctx=ctx, creds=creds)
|
||||||
|
state = await graph.ainvoke(
|
||||||
|
{"messages": [{"role": "user", "content": prompt}]},
|
||||||
|
config={"recursion_limit": DEEPAGENTS_RECURSION_LIMIT},
|
||||||
|
)
|
||||||
|
await ctx.emit_progress("deepagent finished")
|
||||||
|
return _last_message_text(state)
|
||||||
|
|
||||||
|
def _build_deep_agent(
|
||||||
self,
|
self,
|
||||||
ctx: RunContext[PlatformUserAuth],
|
|
||||||
comparison_id: str,
|
|
||||||
quotes: list[QuoteInput],
|
|
||||||
weights: QuoteWeights,
|
|
||||||
) -> dict[str, Any]:
|
|
||||||
tenant = _tenant_key(ctx)
|
|
||||||
validation = _validate_comparison_request(comparison_id, quotes, weights)
|
|
||||||
if validation is not None:
|
|
||||||
return validation
|
|
||||||
|
|
||||||
result = _score_quotes(comparison_id, quotes, weights)
|
|
||||||
receipt = _build_receipt(
|
|
||||||
tenant_key=tenant,
|
|
||||||
skill_name="compare_quotes",
|
|
||||||
comparison_id=comparison_id,
|
|
||||||
inputs={
|
|
||||||
"comparison_id": comparison_id,
|
|
||||||
"quotes": [quote.model_dump(mode="json") for quote in quotes],
|
|
||||||
"weights": weights.model_dump(mode="json"),
|
|
||||||
},
|
|
||||||
result=result,
|
|
||||||
)
|
|
||||||
persisted = await asyncio.to_thread(_persist_comparison, tenant, comparison_id, result, receipt)
|
|
||||||
if not persisted["ok"]:
|
|
||||||
return {
|
|
||||||
"ok": False,
|
|
||||||
"code": persisted["code"],
|
|
||||||
"comparison_id": comparison_id,
|
|
||||||
"receipt": {
|
|
||||||
"receipt_id": receipt["receipt_id"],
|
|
||||||
"persisted": False,
|
|
||||||
"created_at": receipt["created_at"],
|
|
||||||
},
|
|
||||||
}
|
|
||||||
result["receipt"] = {
|
|
||||||
"receipt_id": receipt["receipt_id"],
|
|
||||||
"persisted": True,
|
|
||||||
"created_at": receipt["created_at"],
|
|
||||||
}
|
|
||||||
if persisted.get("warning"):
|
|
||||||
result.setdefault("warnings", []).append(persisted["warning"])
|
|
||||||
await ctx.emit_progress(f"saved comparison {comparison_id}")
|
|
||||||
return result
|
|
||||||
|
|
||||||
@a2a.tool(
|
|
||||||
description="Reopen a saved quote comparison for the authenticated user.",
|
|
||||||
timeout_seconds=20,
|
|
||||||
idempotent=True,
|
|
||||||
cost_class="deterministic",
|
|
||||||
)
|
|
||||||
async def get_comparison(
|
|
||||||
self,
|
|
||||||
ctx: RunContext[PlatformUserAuth],
|
|
||||||
comparison_id: str,
|
|
||||||
) -> dict[str, Any]:
|
|
||||||
tenant = _tenant_key(ctx)
|
|
||||||
if not _valid_comparison_id(comparison_id):
|
|
||||||
return {"ok": False, "code": "invalid_comparison_id", "comparison_id": comparison_id}
|
|
||||||
record = await asyncio.to_thread(_load_comparison, tenant, comparison_id)
|
|
||||||
if record is None:
|
|
||||||
return {"ok": False, "code": "comparison_not_found", "comparison_id": comparison_id}
|
|
||||||
if record.get("ok") is False:
|
|
||||||
return {
|
|
||||||
"ok": False,
|
|
||||||
"code": record.get("code", "persistence_unavailable"),
|
|
||||||
"comparison_id": comparison_id,
|
|
||||||
}
|
|
||||||
result = dict(record["result"])
|
|
||||||
result["comparison_id"] = comparison_id
|
|
||||||
result["ok"] = True
|
|
||||||
result["reloaded"] = True
|
|
||||||
result["saved_at"] = record.get("updated_at")
|
|
||||||
if record.get("latest_receipt_id"):
|
|
||||||
result["latest_receipt_id"] = record["latest_receipt_id"]
|
|
||||||
return result
|
|
||||||
|
|
||||||
@a2a.tool(
|
|
||||||
description="Browser-safe upload bridge: accept bounded base64 JSON/CSV/text quote documents, parse them, compare, persist, and receipt the result.",
|
|
||||||
timeout_seconds=30,
|
|
||||||
idempotent=True,
|
|
||||||
cost_class="deterministic",
|
|
||||||
)
|
|
||||||
async def compare_quotes_from_browser_upload(
|
|
||||||
self,
|
|
||||||
ctx: RunContext[PlatformUserAuth],
|
|
||||||
comparison_id: str,
|
|
||||||
documents: list[BrowserDocument],
|
|
||||||
weights: QuoteWeights,
|
|
||||||
) -> dict[str, Any]:
|
|
||||||
parsed = _decode_browser_documents(documents)
|
|
||||||
if not parsed["ok"]:
|
|
||||||
return {"ok": False, "comparison_id": comparison_id, **parsed}
|
|
||||||
quotes = parsed["quotes"]
|
|
||||||
result = await self.compare_quotes(ctx, comparison_id, quotes, weights)
|
|
||||||
result["source"] = "browser_upload"
|
|
||||||
result["parsed_documents"] = parsed["documents"]
|
|
||||||
return result
|
|
||||||
|
|
||||||
@a2a.tool(
|
|
||||||
description="External Agent API upload path: accept a typed FileUpload JSON/CSV/text document, parse quotes, compare, persist, and receipt the result.",
|
|
||||||
timeout_seconds=30,
|
|
||||||
idempotent=True,
|
|
||||||
cost_class="deterministic",
|
|
||||||
)
|
|
||||||
async def compare_quotes_from_file(
|
|
||||||
self,
|
|
||||||
ctx: RunContext[PlatformUserAuth],
|
|
||||||
comparison_id: str,
|
|
||||||
document: Annotated[
|
|
||||||
UploadedFile,
|
|
||||||
FileUpload(
|
|
||||||
accept=sorted(ALLOWED_UPLOAD_MEDIA_TYPES),
|
|
||||||
max_bytes=MAX_UPLOAD_BYTES,
|
|
||||||
description="JSON or CSV quote file with vendor, quantity, unit_price, delivery_days, warranty_months.",
|
|
||||||
),
|
|
||||||
],
|
|
||||||
weights: QuoteWeights,
|
|
||||||
) -> dict[str, Any]:
|
|
||||||
if document.size_bytes > MAX_UPLOAD_BYTES:
|
|
||||||
return {"ok": False, "code": "file_too_large", "comparison_id": comparison_id}
|
|
||||||
media_type = (document.media_type or "").split(";", 1)[0].strip().lower()
|
|
||||||
if media_type not in ALLOWED_UPLOAD_MEDIA_TYPES:
|
|
||||||
return {"ok": False, "code": "unsupported_media_type", "comparison_id": comparison_id}
|
|
||||||
try:
|
|
||||||
reader = getattr(ctx.workspace, "read_bytes", None)
|
|
||||||
if reader is None:
|
|
||||||
return {"ok": False, "code": "workspace_read_unavailable", "comparison_id": comparison_id}
|
|
||||||
data = reader(document.path)
|
|
||||||
except Exception:
|
|
||||||
return {"ok": False, "code": "file_read_failed", "comparison_id": comparison_id}
|
|
||||||
parsed = _parse_quote_document(data, filename=document.filename, media_type=media_type)
|
|
||||||
if not parsed["ok"]:
|
|
||||||
return {"ok": False, "comparison_id": comparison_id, **parsed}
|
|
||||||
result = await self.compare_quotes(ctx, comparison_id, parsed["quotes"], weights)
|
|
||||||
result["source"] = "file_upload"
|
|
||||||
result["parsed_documents"] = [{"filename": document.filename, "quotes": len(parsed["quotes"])}]
|
|
||||||
return result
|
|
||||||
|
|
||||||
|
|
||||||
def _tenant_key(ctx: RunContext[PlatformUserAuth]) -> str:
|
|
||||||
auth = ctx.auth
|
|
||||||
stable = auth.user_id if auth.user_id is not None else (auth.sub or auth.email)
|
|
||||||
return f"user:{stable}"
|
|
||||||
|
|
||||||
|
|
||||||
def _valid_comparison_id(value: str) -> bool:
|
|
||||||
return bool(COMPARISON_ID_RE.fullmatch(str(value or "").strip()))
|
|
||||||
|
|
||||||
|
|
||||||
def _validate_comparison_request(
|
|
||||||
comparison_id: str,
|
|
||||||
quotes: list[QuoteInput],
|
|
||||||
weights: QuoteWeights,
|
|
||||||
) -> dict[str, Any] | None:
|
|
||||||
if not _valid_comparison_id(comparison_id):
|
|
||||||
return {"ok": False, "code": "invalid_comparison_id", "comparison_id": comparison_id}
|
|
||||||
if len(quotes) < 2:
|
|
||||||
return {"ok": False, "code": "at_least_two_quotes_required", "comparison_id": comparison_id}
|
|
||||||
if len(quotes) > MAX_QUOTES:
|
|
||||||
return {"ok": False, "code": "too_many_quotes", "comparison_id": comparison_id}
|
|
||||||
vendor_names = [quote.vendor.casefold() for quote in quotes]
|
|
||||||
if len(set(vendor_names)) != len(vendor_names):
|
|
||||||
return {"ok": False, "code": "duplicate_vendor", "comparison_id": comparison_id}
|
|
||||||
if weights.price + weights.delivery + weights.warranty <= 0:
|
|
||||||
return {"ok": False, "code": "weights_sum_to_zero", "comparison_id": comparison_id}
|
|
||||||
return None
|
|
||||||
|
|
||||||
|
|
||||||
def _rank_sort_key(row: dict[str, Any]) -> tuple[Any, ...]:
|
|
||||||
return (-row["score"], row["total_price"], row["delivery_days"], -row["warranty_months"], row["vendor"])
|
|
||||||
|
|
||||||
|
|
||||||
def _score_quotes(comparison_id: str, quotes: list[QuoteInput], weights: QuoteWeights) -> dict[str, Any]:
|
|
||||||
total_weight = Decimal(str(weights.price + weights.delivery + weights.warranty))
|
|
||||||
prices = [_money(q.unit_price) for q in quotes]
|
|
||||||
totals = [_money(q.unit_price * q.quantity) for q in quotes]
|
|
||||||
deliveries = [Decimal(q.delivery_days) for q in quotes]
|
|
||||||
warranties = [Decimal(q.warranty_months) for q in quotes]
|
|
||||||
|
|
||||||
rows: list[dict[str, Any]] = []
|
|
||||||
for index, quote in enumerate(quotes):
|
|
||||||
price_score = _lower_is_better(totals[index], totals)
|
|
||||||
delivery_score = _lower_is_better(deliveries[index], deliveries)
|
|
||||||
warranty_score = _higher_is_better(warranties[index], warranties)
|
|
||||||
weighted_score = (
|
|
||||||
price_score * Decimal(str(weights.price))
|
|
||||||
+ delivery_score * Decimal(str(weights.delivery))
|
|
||||||
+ warranty_score * Decimal(str(weights.warranty))
|
|
||||||
) / total_weight
|
|
||||||
rows.append(
|
|
||||||
{
|
|
||||||
"vendor": quote.vendor,
|
|
||||||
"quantity": quote.quantity,
|
|
||||||
"unit_price": _float2(prices[index]),
|
|
||||||
"total_price": _float2(totals[index]),
|
|
||||||
"delivery_days": quote.delivery_days,
|
|
||||||
"warranty_months": quote.warranty_months,
|
|
||||||
"score": _float2(weighted_score),
|
|
||||||
"score_breakdown": {
|
|
||||||
"price": _float2(price_score),
|
|
||||||
"delivery": _float2(delivery_score),
|
|
||||||
"warranty": _float2(warranty_score),
|
|
||||||
},
|
|
||||||
}
|
|
||||||
)
|
|
||||||
rows.sort(key=_rank_sort_key)
|
|
||||||
for rank, row in enumerate(rows, start=1):
|
|
||||||
row["rank"] = rank
|
|
||||||
recommendation = rows[0]
|
|
||||||
return {
|
|
||||||
"ok": True,
|
|
||||||
"comparison_id": comparison_id,
|
|
||||||
"recommendation": {
|
|
||||||
"vendor": recommendation["vendor"],
|
|
||||||
"score": recommendation["score"],
|
|
||||||
"total_price": recommendation["total_price"],
|
|
||||||
"delivery_days": recommendation["delivery_days"],
|
|
||||||
"warranty_months": recommendation["warranty_months"],
|
|
||||||
"rationale": (
|
|
||||||
f"{recommendation['vendor']} has the best weighted score using "
|
|
||||||
f"price={weights.price:g}, delivery={weights.delivery:g}, warranty={weights.warranty:g}."
|
|
||||||
),
|
|
||||||
},
|
|
||||||
"rankings": rows,
|
|
||||||
"weights": weights.model_dump(mode="json"),
|
|
||||||
"quote_count": len(quotes),
|
|
||||||
}
|
|
||||||
|
|
||||||
|
|
||||||
def _money(value: float | int | Decimal) -> Decimal:
|
|
||||||
return Decimal(str(value)).quantize(Decimal("0.01"), rounding=ROUND_HALF_UP)
|
|
||||||
|
|
||||||
|
|
||||||
def _float2(value: Decimal) -> float:
|
|
||||||
return float(value.quantize(Decimal("0.01"), rounding=ROUND_HALF_UP))
|
|
||||||
|
|
||||||
|
|
||||||
def _lower_is_better(value: Decimal, values: list[Decimal]) -> Decimal:
|
|
||||||
low, high = min(values), max(values)
|
|
||||||
if low == high:
|
|
||||||
return Decimal("100")
|
|
||||||
return ((high - value) / (high - low) * Decimal("100")).quantize(Decimal("0.01"), rounding=ROUND_HALF_UP)
|
|
||||||
|
|
||||||
|
|
||||||
def _higher_is_better(value: Decimal, values: list[Decimal]) -> Decimal:
|
|
||||||
low, high = min(values), max(values)
|
|
||||||
if low == high:
|
|
||||||
return Decimal("100")
|
|
||||||
return ((value - low) / (high - low) * Decimal("100")).quantize(Decimal("0.01"), rounding=ROUND_HALF_UP)
|
|
||||||
|
|
||||||
|
|
||||||
def _decode_browser_documents(documents: list[BrowserDocument]) -> dict[str, Any]:
|
|
||||||
if not documents:
|
|
||||||
return {"ok": False, "code": "no_documents_provided"}
|
|
||||||
if len(documents) > MAX_BROWSER_DOCUMENTS:
|
|
||||||
return {"ok": False, "code": "too_many_documents"}
|
|
||||||
all_quotes: list[QuoteInput] = []
|
|
||||||
summaries: list[dict[str, Any]] = []
|
|
||||||
for doc in documents:
|
|
||||||
try:
|
|
||||||
data = base64.b64decode(doc.data_base64.encode("ascii"), validate=True)
|
|
||||||
except Exception:
|
|
||||||
return {"ok": False, "code": "invalid_base64", "filename": doc.filename}
|
|
||||||
if len(data) > MAX_UPLOAD_BYTES:
|
|
||||||
return {"ok": False, "code": "file_too_large", "filename": doc.filename}
|
|
||||||
parsed = _parse_quote_document(data, filename=doc.filename, media_type=doc.media_type)
|
|
||||||
if not parsed["ok"]:
|
|
||||||
return {"ok": False, "filename": doc.filename, **parsed}
|
|
||||||
all_quotes.extend(parsed["quotes"])
|
|
||||||
summaries.append({"filename": doc.filename, "quotes": len(parsed["quotes"])})
|
|
||||||
return {"ok": True, "quotes": all_quotes, "documents": summaries}
|
|
||||||
|
|
||||||
|
|
||||||
def _parse_quote_document(data: bytes, *, filename: str, media_type: str) -> dict[str, Any]:
|
|
||||||
if len(data) > MAX_UPLOAD_BYTES:
|
|
||||||
return {"ok": False, "code": "file_too_large"}
|
|
||||||
text = data.decode("utf-8-sig", errors="replace").strip()
|
|
||||||
if not text:
|
|
||||||
return {"ok": False, "code": "empty_document"}
|
|
||||||
try:
|
|
||||||
if media_type in {"application/json", "text/json"} or filename.lower().endswith(".json"):
|
|
||||||
raw = json.loads(text)
|
|
||||||
items = raw.get("quotes", raw) if isinstance(raw, dict) else raw
|
|
||||||
if not isinstance(items, list):
|
|
||||||
return {"ok": False, "code": "quotes_array_required"}
|
|
||||||
return {"ok": True, "quotes": [QuoteInput.model_validate(item) for item in items]}
|
|
||||||
rows = list(csv.DictReader(io.StringIO(text)))
|
|
||||||
if rows and rows[0]:
|
|
||||||
return {"ok": True, "quotes": [_quote_from_row(row) for row in rows]}
|
|
||||||
return {"ok": False, "code": "unsupported_document_format"}
|
|
||||||
except Exception:
|
|
||||||
return {"ok": False, "code": "document_parse_failed"}
|
|
||||||
|
|
||||||
|
|
||||||
def _quote_from_row(row: dict[str, Any]) -> QuoteInput:
|
|
||||||
normalized = {str(key).strip().lower(): value for key, value in row.items()}
|
|
||||||
return QuoteInput(
|
|
||||||
vendor=str(normalized.get("vendor") or normalized.get("supplier") or ""),
|
|
||||||
quantity=int(float(normalized.get("quantity") or normalized.get("qty") or 0)),
|
|
||||||
unit_price=float(normalized.get("unit_price") or normalized.get("price") or 0),
|
|
||||||
delivery_days=int(float(normalized.get("delivery_days") or normalized.get("delivery") or 0)),
|
|
||||||
warranty_months=int(float(normalized.get("warranty_months") or normalized.get("warranty") or 0)),
|
|
||||||
)
|
|
||||||
|
|
||||||
|
|
||||||
def _database_url() -> str | None:
|
|
||||||
return os.environ.get("DATABASE_URL")
|
|
||||||
|
|
||||||
|
|
||||||
def _connect() -> Any:
|
|
||||||
if psycopg is None:
|
|
||||||
raise RuntimeError("psycopg is not installed")
|
|
||||||
url = _database_url()
|
|
||||||
if not url:
|
|
||||||
raise RuntimeError("DATABASE_URL is not configured")
|
|
||||||
return psycopg.connect(
|
|
||||||
url,
|
|
||||||
row_factory=dict_row,
|
|
||||||
connect_timeout=DB_CONNECT_TIMEOUT_SECONDS,
|
|
||||||
options=(
|
|
||||||
f"-c statement_timeout={DB_STATEMENT_TIMEOUT_MS} "
|
|
||||||
f"-c lock_timeout={DB_LOCK_TIMEOUT_MS}"
|
|
||||||
),
|
|
||||||
)
|
|
||||||
|
|
||||||
|
|
||||||
def _persist_comparison(
|
|
||||||
tenant_key: str,
|
|
||||||
comparison_id: str,
|
|
||||||
result: dict[str, Any],
|
|
||||||
receipt: dict[str, Any],
|
|
||||||
) -> dict[str, Any]:
|
|
||||||
if not _database_url() or psycopg is None:
|
|
||||||
return {"ok": False, "code": "persistence_unavailable"}
|
|
||||||
try:
|
|
||||||
with _connect() as conn:
|
|
||||||
with conn.cursor() as cur:
|
|
||||||
cur.execute(
|
|
||||||
"""
|
|
||||||
INSERT INTO quote_judge_comparisons
|
|
||||||
(tenant_key, comparison_id, input_json, result_json, recommendation_vendor, latest_receipt_id, updated_at)
|
|
||||||
VALUES (%s, %s, %s, %s, %s, %s, NOW())
|
|
||||||
ON CONFLICT (tenant_key, comparison_id) DO UPDATE SET
|
|
||||||
input_json = EXCLUDED.input_json,
|
|
||||||
result_json = EXCLUDED.result_json,
|
|
||||||
recommendation_vendor = EXCLUDED.recommendation_vendor,
|
|
||||||
latest_receipt_id = EXCLUDED.latest_receipt_id,
|
|
||||||
updated_at = NOW()
|
|
||||||
""",
|
|
||||||
(
|
|
||||||
tenant_key,
|
|
||||||
comparison_id,
|
|
||||||
Jsonb({"weights": result["weights"], "quote_count": result["quote_count"]}),
|
|
||||||
Jsonb(result),
|
|
||||||
result["recommendation"]["vendor"],
|
|
||||||
receipt["receipt_id"],
|
|
||||||
),
|
|
||||||
)
|
|
||||||
cur.execute(
|
|
||||||
"""
|
|
||||||
INSERT INTO quote_judge_receipts
|
|
||||||
(receipt_id, tenant_key, comparison_id, skill_name, tool_name, status, input_hash, result_hash, receipt_json, payload)
|
|
||||||
VALUES (%s, %s, %s, %s, %s, %s, %s, %s, %s, %s)
|
|
||||||
ON CONFLICT (receipt_id) DO UPDATE SET
|
|
||||||
tenant_key = EXCLUDED.tenant_key,
|
|
||||||
comparison_id = EXCLUDED.comparison_id,
|
|
||||||
skill_name = EXCLUDED.skill_name,
|
|
||||||
tool_name = EXCLUDED.tool_name,
|
|
||||||
status = EXCLUDED.status,
|
|
||||||
input_hash = EXCLUDED.input_hash,
|
|
||||||
result_hash = EXCLUDED.result_hash,
|
|
||||||
receipt_json = EXCLUDED.receipt_json,
|
|
||||||
payload = EXCLUDED.payload
|
|
||||||
""",
|
|
||||||
(
|
|
||||||
receipt["receipt_id"],
|
|
||||||
tenant_key,
|
|
||||||
comparison_id,
|
|
||||||
receipt["skill_name"],
|
|
||||||
receipt["skill_name"],
|
|
||||||
receipt["status"],
|
|
||||||
receipt["input_hash"],
|
|
||||||
receipt["result_hash"],
|
|
||||||
Jsonb(receipt),
|
|
||||||
Jsonb(receipt),
|
|
||||||
),
|
|
||||||
)
|
|
||||||
conn.commit()
|
|
||||||
return {"ok": True}
|
|
||||||
except Exception:
|
|
||||||
return {"ok": False, "code": "persistence_unavailable"}
|
|
||||||
|
|
||||||
|
|
||||||
def _load_comparison(tenant_key: str, comparison_id: str) -> dict[str, Any] | None:
|
|
||||||
if not _database_url() or psycopg is None:
|
|
||||||
return {"ok": False, "code": "persistence_unavailable"}
|
|
||||||
try:
|
|
||||||
with _connect() as conn:
|
|
||||||
with conn.cursor() as cur:
|
|
||||||
cur.execute(
|
|
||||||
"""
|
|
||||||
SELECT result_json, updated_at, latest_receipt_id
|
|
||||||
FROM quote_judge_comparisons
|
|
||||||
WHERE tenant_key = %s AND comparison_id = %s
|
|
||||||
""",
|
|
||||||
(tenant_key, comparison_id),
|
|
||||||
)
|
|
||||||
row = cur.fetchone()
|
|
||||||
if not row:
|
|
||||||
return None
|
|
||||||
updated_at = row["updated_at"]
|
|
||||||
return {
|
|
||||||
"result": row["result_json"],
|
|
||||||
"updated_at": updated_at.isoformat() if hasattr(updated_at, "isoformat") else str(updated_at),
|
|
||||||
"latest_receipt_id": row.get("latest_receipt_id"),
|
|
||||||
}
|
|
||||||
except Exception:
|
|
||||||
return {"ok": False, "code": "persistence_unavailable"}
|
|
||||||
|
|
||||||
|
|
||||||
def _build_receipt(
|
|
||||||
*,
|
*,
|
||||||
tenant_key: str,
|
ctx: RunContext[{{ auth_type }}],
|
||||||
skill_name: str,
|
creds: LLMCreds,
|
||||||
comparison_id: str,
|
) -> Any:
|
||||||
inputs: dict[str, Any],
|
# Lazy imports keep `a2a card` usable before local dependencies are
|
||||||
result: dict[str, Any],
|
# installed. `a2a deploy` installs requirements.txt during the build.
|
||||||
) -> dict[str, Any]:
|
from a2a_pack.deepagents import create_a2a_deep_agent
|
||||||
now = datetime.now(timezone.utc).isoformat()
|
from langchain.agents.middleware import wrap_model_call
|
||||||
tenant_hash = _hash_json({"tenant_key": tenant_key})
|
from langchain_core.tools import tool
|
||||||
input_hash = _hash_json(inputs)
|
|
||||||
result_hash = _hash_json(result)
|
@tool
|
||||||
receipt_id = "qjr_" + _hash_json({
|
def text_stats(text: str) -> str:
|
||||||
"tenant_hash": tenant_hash,
|
"""Return exact word, character, and line counts for text."""
|
||||||
"skill_name": skill_name,
|
words = [part for part in text.split() if part.strip()]
|
||||||
"comparison_id": comparison_id,
|
return json.dumps(
|
||||||
"input_hash": input_hash,
|
{
|
||||||
"result_hash": result_hash,
|
"characters": len(text),
|
||||||
})[:24]
|
"words": len(words),
|
||||||
return {
|
"lines": len(text.splitlines()) or 1,
|
||||||
"schema": "quotejudge.production_receipt.v1",
|
|
||||||
"receipt_id": receipt_id,
|
|
||||||
"agent_name": "quote-judge-studio-v1",
|
|
||||||
"agent_version": AGENT_VERSION,
|
|
||||||
"skill_name": skill_name,
|
|
||||||
"comparison_id": comparison_id,
|
|
||||||
"status": "ok" if result.get("ok") else "error",
|
|
||||||
"tenant_hash": tenant_hash,
|
|
||||||
"input_hash": input_hash,
|
|
||||||
"result_hash": result_hash,
|
|
||||||
"created_at": now,
|
|
||||||
}
|
}
|
||||||
|
)
|
||||||
|
|
||||||
|
@wrap_model_call
|
||||||
|
async def log_model_call(request: Any, handler: Any) -> Any:
|
||||||
|
messages = request.state.get("messages", [])
|
||||||
|
print(
|
||||||
|
"[middleware] model_call "
|
||||||
|
f"model={creds.model} source={creds.source} messages={len(messages)}"
|
||||||
|
)
|
||||||
|
return await handler(request)
|
||||||
|
|
||||||
|
backend = ctx.workspace_backend()
|
||||||
|
skill_sources = _seed_runtime_skills(backend, ctx)
|
||||||
|
# create_a2a_deep_agent resolves provider:model strings with
|
||||||
|
# langchain.init_chat_model from ctx.llm, preserving LiteLLM routing,
|
||||||
|
# provider-specific extra body, and runtime model overrides.
|
||||||
|
return create_a2a_deep_agent(
|
||||||
|
ctx,
|
||||||
|
creds=creds,
|
||||||
|
backend=backend,
|
||||||
|
skills=skill_sources or None,
|
||||||
|
tools=[text_stats],
|
||||||
|
middleware=[log_model_call],
|
||||||
|
system_prompt=SYSTEM_PROMPT,
|
||||||
|
)
|
||||||
|
|
||||||
|
|
||||||
def _hash_json(value: Any) -> str:
|
def _runtime_skills_root(ctx: RunContext[Any]) -> str:
|
||||||
encoded = json.dumps(value, sort_keys=True, separators=(",", ":"), default=str).encode("utf-8")
|
workspace = getattr(ctx, "_workspace", None)
|
||||||
return hashlib.sha256(encoded).hexdigest()
|
prefixes = tuple(getattr(workspace, "write_prefixes", ()) or ())
|
||||||
|
if not prefixes:
|
||||||
|
outputs_prefix = getattr(workspace, "outputs_prefix", None)
|
||||||
|
prefixes = (outputs_prefix or "outputs/",)
|
||||||
|
prefix = str(prefixes[0]).strip("/")
|
||||||
|
return f"/{prefix}/{RUNTIME_SKILLS_DIR}" if prefix else f"/{RUNTIME_SKILLS_DIR}"
|
||||||
|
|
||||||
|
|
||||||
|
def _seed_runtime_skills(backend: Any, ctx: RunContext[Any]) -> list[str]:
|
||||||
|
"""Copy packaged DeepAgents skills into the invocation workspace.
|
||||||
|
|
||||||
|
DeepAgents loads skills from its backend, while source-controlled
|
||||||
|
``skills/`` folders live in the image. This bridge lets generated agents
|
||||||
|
ship reusable SKILL.md bundles without giving up durable A2A workspace
|
||||||
|
files.
|
||||||
|
"""
|
||||||
|
root = Path(__file__).parent / "skills"
|
||||||
|
if not root.exists():
|
||||||
|
return []
|
||||||
|
runtime_skills_root = _runtime_skills_root(ctx)
|
||||||
|
uploads: list[tuple[str, bytes]] = []
|
||||||
|
for path in root.rglob("*"):
|
||||||
|
if path.is_file():
|
||||||
|
rel = path.relative_to(root).as_posix()
|
||||||
|
uploads.append((runtime_skills_root + rel, path.read_bytes()))
|
||||||
|
if uploads:
|
||||||
|
backend.upload_files(uploads)
|
||||||
|
return [runtime_skills_root]
|
||||||
|
return []
|
||||||
|
|
||||||
|
|
||||||
|
def _last_message_text(state: dict[str, Any]) -> str:
|
||||||
|
messages = state.get("messages") or []
|
||||||
|
if not messages:
|
||||||
|
return json.dumps(state, default=str)
|
||||||
|
|
||||||
|
content = getattr(messages[-1], "content", None)
|
||||||
|
if isinstance(content, str):
|
||||||
|
return content
|
||||||
|
if isinstance(content, list):
|
||||||
|
parts: list[str] = []
|
||||||
|
for item in content:
|
||||||
|
if isinstance(item, dict):
|
||||||
|
text = item.get("text") or item.get("content")
|
||||||
|
if text:
|
||||||
|
parts.append(str(text))
|
||||||
|
elif item:
|
||||||
|
parts.append(str(item))
|
||||||
|
return "\n".join(parts) if parts else json.dumps(content, default=str)
|
||||||
|
return str(content or messages[-1])
|
||||||
|
|||||||
Reference in New Issue
Block a user