proposal-system/lambdas/suggestions/app.py
Adam Moussa 9d6612e337 Implement Lambda functions for PDF processing, suggestions, and library ingest
- pdf-extract: Parse vendor PDFs with pdfplumber, fallback to Claude multimodal
- pdf-generate: Generate branded proposal PDFs with reportlab Platypus
- library-ingest: Format approved proposals as markdown and sync to Bedrock KB
- suggestions: Query KB for similar proposals, generate line items via Claude
- All Lambdas use internal API key auth and cold-start secret caching
- Fix pdf_path unbound variable in pdf-extract error handling
2026-05-16 22:10:21 -04:00

301 lines
9.6 KiB
Python

"""Proposal System - Suggestion Engine Lambda.
Queries Bedrock Knowledge Base for similar proposals and invokes Claude
to generate line item suggestions for new proposals.
"""
import json
import os
import boto3
import httpx
KNOWLEDGE_BASE_ID = os.environ.get("KNOWLEDGE_BASE_ID", "")
MODEL_ID = os.environ.get("MODEL_ID", "us.anthropic.claude-sonnet-4-5-20250929-v1:0")
API_BASE_URL = os.environ.get("API_BASE_URL", "")
INTERNAL_API_KEY_SECRET_ARN = os.environ.get("INTERNAL_API_KEY_SECRET_ARN", "")
bedrock_agent = boto3.client("bedrock-agent-runtime")
bedrock_runtime = boto3.client("bedrock-runtime")
secrets_client = boto3.client("secretsmanager")
_cached_api_key: str | None = None
def _get_api_key() -> str:
global _cached_api_key
if _cached_api_key is None:
if INTERNAL_API_KEY_SECRET_ARN:
resp = secrets_client.get_secret_value(SecretId=INTERNAL_API_KEY_SECRET_ARN)
_cached_api_key = resp["SecretString"]
else:
_cached_api_key = ""
return _cached_api_key
def handler(event, context):
for record in event.get("Records", []):
body = json.loads(record["body"])
payload = body.get("payload", body)
proposal_id = payload["proposalId"]
trigger = payload.get("trigger", "generate")
process_suggestion(proposal_id, trigger)
return {"statusCode": 200}
def process_suggestion(proposal_id: str, trigger: str):
proposal = fetch_proposal(proposal_id)
if not proposal:
print(f"Proposal {proposal_id} not found")
return
scope = proposal.get("refinedScope") or proposal.get("scopeOfWork", "")
category = proposal.get("serviceCategory", "")
priority = proposal.get("priority", "")
existing_items = fetch_line_items(proposal_id)
similar_proposals = retrieve_similar(scope, category)
suggested_items = generate_line_items(scope, category, priority, similar_proposals)
post_line_items(proposal_id, suggested_items, existing_items)
store_similar_references(proposal_id, similar_proposals)
update_status_to_in_review(proposal_id)
def fetch_line_items(proposal_id: str) -> list[dict]:
try:
resp = httpx.get(
f"{API_BASE_URL}/api/proposals/{proposal_id}/line-items",
headers=_api_headers(),
timeout=10,
)
if resp.status_code == 200:
return resp.json()
except Exception as e:
print(f"Error fetching line items: {e}")
return []
def fetch_proposal(proposal_id: str) -> dict | None:
try:
resp = httpx.get(
f"{API_BASE_URL}/api/proposals/{proposal_id}",
headers=_api_headers(),
timeout=10,
)
if resp.status_code == 200:
return resp.json()
except Exception as e:
print(f"Error fetching proposal: {e}")
return None
def retrieve_similar(scope: str, category: str) -> list[dict]:
if not KNOWLEDGE_BASE_ID:
print("No Knowledge Base configured, skipping retrieval")
return []
try:
filter_config = {
"equals": {"key": "service_category", "value": category}
} if category else None
params = {
"knowledgeBaseId": KNOWLEDGE_BASE_ID,
"retrievalQuery": {"text": scope},
"retrievalConfiguration": {
"vectorSearchConfiguration": {
"numberOfResults": 10,
}
},
}
if filter_config:
params["retrievalConfiguration"]["vectorSearchConfiguration"]["filter"] = filter_config
response = bedrock_agent.retrieve(**params)
results = []
for result in response.get("retrievalResults", []):
content = result.get("content", {}).get("text", "")
score = result.get("score", 0.0)
metadata = result.get("metadata", {})
source_uri = result.get("location", {}).get("s3Location", {}).get("uri", "")
results.append({
"content": content,
"score": score,
"metadata": metadata,
"sourceUri": source_uri,
})
return results
except Exception as e:
print(f"Error retrieving from KB: {e}")
return []
def generate_line_items(
scope: str,
category: str,
priority: str,
similar_proposals: list[dict],
) -> list[dict]:
context_block = ""
if similar_proposals:
context_block = "Here are similar historical proposals and their line items for reference:\n\n"
for i, sp in enumerate(similar_proposals[:5], 1):
context_block += f"--- Similar Proposal {i} (relevance: {sp['score']:.2f}) ---\n"
context_block += sp["content"] + "\n\n"
prompt = f"""You are a construction/facilities proposal estimator for Sea Haven Industries.
Based on the scope of work and similar historical proposals, generate a detailed list of line items
with quantities, units, and estimated pricing.
Service Category: {category}
Priority: {priority}
Scope of Work:
{scope}
{context_block}
Generate line items as a JSON array. Each item should have:
- description: clear description of the work/material
- quantity: numeric quantity
- unit: unit of measurement (e.g., "sq ft", "hours", "each", "linear ft")
- unitPrice: price per unit in dollars (or null if lump sum)
- totalPrice: total price for this line item in dollars
- pricingMode: "UnitPrice" if unit price provided, "TotalPrice" if lump sum
Respond ONLY with the JSON array, no additional text."""
try:
response = bedrock_runtime.invoke_model(
modelId=MODEL_ID,
contentType="application/json",
accept="application/json",
body=json.dumps({
"anthropic_version": "bedrock-2023-05-31",
"max_tokens": 4096,
"messages": [{"role": "user", "content": prompt}],
"temperature": 0.3,
}),
)
response_body = json.loads(response["body"].read())
content = response_body["content"][0]["text"]
content = content.strip()
if content.startswith("```"):
content = content.split("\n", 1)[1]
content = content.rsplit("```", 1)[0]
line_items = json.loads(content)
return line_items if isinstance(line_items, list) else []
except Exception as e:
print(f"Error generating line items: {e}")
return []
def post_line_items(proposal_id: str, items: list[dict], existing_items: list[dict]):
if not items and not existing_items:
return
line_items_payload = []
# Preserve non-AI items (Manual, Vendor, Historical)
preserved = [li for li in existing_items if li.get("source") != "AI"]
for i, li in enumerate(preserved):
line_items_payload.append({
"id": li.get("id"),
"description": li["description"],
"quantity": float(li.get("quantity", 1)),
"unit": li.get("unit", "each"),
"unitPrice": li.get("unitPrice"),
"totalPrice": float(li.get("totalPrice", 0)),
"pricingMode": li.get("pricingMode", "TotalPrice"),
"sortOrder": i + 1,
"source": li.get("source", "Manual"),
})
# Add new AI-generated items after preserved ones
offset = len(line_items_payload)
for i, item in enumerate(items):
pricing_mode = item.get("pricingMode", "TotalPrice")
if pricing_mode not in ("UnitPrice", "TotalPrice", "Both"):
pricing_mode = "UnitPrice" if item.get("unitPrice") else "TotalPrice"
line_items_payload.append({
"id": None,
"description": item["description"],
"quantity": float(item.get("quantity", 1)),
"unit": item.get("unit", "each"),
"unitPrice": item.get("unitPrice"),
"totalPrice": float(item.get("totalPrice", 0)),
"pricingMode": pricing_mode,
"sortOrder": offset + i + 1,
"source": "AI",
})
try:
resp = httpx.put(
f"{API_BASE_URL}/api/proposals/{proposal_id}/line-items",
json={"lineItems": line_items_payload},
headers=_api_headers(),
timeout=15,
)
if resp.status_code not in (200, 201):
print(f"Failed to post line items: {resp.status_code} {resp.text}")
except Exception as e:
print(f"Error posting line items: {e}")
def store_similar_references(proposal_id: str, similar_proposals: list[dict]):
if not similar_proposals:
return
for sp in similar_proposals[:5]:
source_uri = sp.get("sourceUri", "")
library_item_id = source_uri.split("/")[-1] if source_uri else ""
if not library_item_id:
continue
try:
httpx.post(
f"{API_BASE_URL}/api/proposals/{proposal_id}/similar-references",
json={
"referencedLibraryItemId": library_item_id,
"similarityScore": sp["score"],
},
headers=_api_headers(),
timeout=10,
)
except Exception as e:
print(f"Error storing similar reference: {e}")
def update_status_to_in_review(proposal_id: str):
try:
httpx.put(
f"{API_BASE_URL}/api/proposals/{proposal_id}",
json={"status": "InReview"},
headers=_api_headers(),
timeout=10,
)
except Exception as e:
print(f"Error updating status: {e}")
def _api_headers() -> dict:
headers = {"Content-Type": "application/json"}
api_key = _get_api_key()
if api_key:
headers["X-Internal-Api-Key"] = api_key
return headers