Capstone 3 — Domain C: Entity Resolution Agent

mkdir capstone-3-entity-resolution && cd capstone-3-entity-resolution
python -m venv venv && source venv/bin/activate   # Windows: venv\Scripts\Activate.ps1

# Pin dependencies for reproducibility
echo "anthropic>=0.40.0" > requirements.txt
pip install -r requirements.txt
export ANTHROPIC_API_KEY=your-key-here             # Windows: set ANTHROPIC_API_KEY=your-key-here

capstone-3-entity-resolution/
├── entity_tools.py      # Fuzzy matching + mock data tools
├── entity_agent.py      # Entity resolution agent with ReAct loop
├── entity_tools.ts      # TypeScript tools
├── entity_agent.ts      # TypeScript agent
└── requirements.txt     # Dependencies

{
  "request_id": "ER-2024-0078",
  "input_name": "Acme Logistics LLC",
  "input_state": "DE",
  "candidates": [
    {"name": "ACME LOGISTICS, L.L.C.", "state": "DE", "filing_count": 3},
    {"name": "Acme Logistics Company", "state": "DE", "filing_count": 1},
    {"name": "Acme Logistics Inc.", "state": "NY", "filing_count": 2}
  ]
}

# entity_tools.py — Mock tools for entity resolution
import re

# --- Tool 1: Search filings by name ---
MOCK_CANDIDATES = {
    "acme logistics llc": [
        {"name": "ACME LOGISTICS, L.L.C.", "state": "DE", "filing_count": 3, "most_recent": "2024-01-15"},
        {"name": "Acme Logistics Company", "state": "DE", "filing_count": 1, "most_recent": "2023-08-20"},
        {"name": "Acme Logistics Inc.", "state": "NY", "filing_count": 2, "most_recent": "2023-12-01"},
    ],
    "buildright construction": [
        {"name": "BuildRight Construction LLC", "state": "NY", "filing_count": 1, "most_recent": "2024-01-10"},
        {"name": "Build Right Construction", "state": "NY", "filing_count": 1, "most_recent": "2022-03-15"},
    ],
}

def search_filings_by_name(business_name: str, state: str | None = None, match_type: str = "fuzzy") -> dict:
    try:
        key = business_name.lower().strip()
        candidates = MOCK_CANDIDATES.get(key, [])
        if state:
            candidates = [c for c in candidates if c["state"] == state.upper()]
        if not candidates:
            return {"is_error": True, "error_category": "NO_RESULTS", "is_retryable": False,
                    "context": f"No filings found for '{business_name}'"}
        return {"is_error": False, "candidates": candidates, "total": len(candidates)}
    except Exception as e:
        return {"is_error": True, "error_category": "INTERNAL_ERROR", "is_retryable": True, "context": str(e)}


# --- Tool 2: Fuzzy match score ---
# NOTE: punctuation is replaced with spaces BEFORE this regex runs, so
# "L.L.C." becomes "l l c" — match that form (with spaces), not "l.l.c."
_SUFFIX_RE = re.compile(
    r'\b(llc|l l c|inc|incorporated|corp|corporation|company|co|ltd)\b'
)

def _normalize(name: str) -> str:
    name = name.lower().strip()
    name = re.sub(r'[,.\-]', ' ', name)
    name = re.sub(r'\s+', ' ', name)
    # Strip ONLY whole-word suffixes (not substrings — "Coca" must
    # not lose "co", "Incorporated" must outrank "corp")
    name = _SUFFIX_RE.sub('', name)
    name = re.sub(r'\s+', ' ', name).strip()
    return name

def fuzzy_match_score(entity_a: str, entity_b: str) -> dict:
    try:
        if not entity_a or not entity_b:
            return {"is_error": True, "error_category": "EMPTY_INPUT", "is_retryable": False, "context": "Both entities required"}
        norm_a, norm_b = _normalize(entity_a), _normalize(entity_b)
        exact = 1.0 if entity_a.lower() == entity_b.lower() else round(len(set(entity_a.lower()) & set(entity_b.lower())) / max(len(set(entity_a.lower())), 1), 2)
        normalized = 1.0 if norm_a == norm_b else round(len(set(norm_a.split()) & set(norm_b.split())) / max(len(set(norm_a.split()) | set(norm_b.split())), 1), 2)
        token_sort = round(normalized * 1.03, 2) if normalized > 0.8 else normalized  # boost for high overlap
        token_sort = min(token_sort, 1.0)
        avg = round((exact + normalized + token_sort) / 3, 2)
        if avg >= 0.85: rec = "likely_match"
        elif avg >= 0.65: rec = "possible_match"
        else: rec = "unlikely_match"
        return {"is_error": False, "entity_a": entity_a, "entity_b": entity_b,
                "scores": {"exact": exact, "normalized": normalized, "token_sort_ratio": token_sort},
                "recommendation": rec}
    except Exception as e:
        return {"is_error": True, "error_category": "INTERNAL_ERROR", "is_retryable": True, "context": str(e)}


# --- Tool 3: Get filing details ---
MOCK_FILINGS = {
    ("acme logistics, l.l.c.", "DE"): {"filings": [
        {"filing_number": "2023-1234567", "secured_party": "First National Bank", "collateral": "All inventory and equipment", "status": "active", "estimated_amount": 750_000},
        {"filing_number": "2022-9876543", "secured_party": "Delaware Capital Partners", "collateral": "All vehicles", "status": "active", "estimated_amount": 350_000},
        {"filing_number": "2024-0011223", "secured_party": "First National Bank", "collateral": "Accounts receivable", "status": "active", "estimated_amount": 1_200_000},
    ]},
}

def get_filing_details(business_name: str, state: str) -> dict:
    try:
        key = (business_name.lower().strip(), state.upper())
        result = MOCK_FILINGS.get(key)
        if result:
            return {"is_error": False, **result}
        return {"is_error": False, "filings": []}
    except Exception as e:
        return {"is_error": True, "error_category": "INTERNAL_ERROR", "is_retryable": True, "context": str(e)}


# --- Tool 4: Business registry ---
MOCK_REGISTRY = {
    ("acme logistics llc", "DE"): {
        "entity_name": "Acme Logistics LLC", "state": "DE", "entity_type": "LLC",
        "file_number": "DE-LLC-2019-4567890", "formation_date": "2019-03-15",
        "status": "active", "principal_address": "456 Commerce Blvd, Dover, DE 19901",
    },
    ("acme logistics inc.", "NY"): {
        "entity_name": "Acme Logistics Inc.", "state": "NY", "entity_type": "Corporation",
        "file_number": "NY-CORP-2020-1234567", "formation_date": "2020-07-01",
        "status": "active", "principal_address": "100 Broadway, New York, NY 10001",
    },
}

def get_business_registry_data(business_name: str, state: str) -> dict:
    try:
        key = (business_name.lower().strip(), state.upper())
        result = MOCK_REGISTRY.get(key)
        if result:
            return {"is_error": False, **result}
        return {"is_error": True, "error_category": "NOT_FOUND", "is_retryable": False,
                "context": f"No registry entry for '{business_name}' in {state}"}
    except Exception as e:
        return {"is_error": True, "error_category": "INTERNAL_ERROR", "is_retryable": True, "context": str(e)}


# --- Tool 5: Merge entity profile ---
def merge_entity_profile(primary_entity: dict, merge_candidates: list, confidence: float) -> dict:
    try:
        if confidence < 0.5:
            return {"is_error": True, "error_category": "INSUFFICIENT_EVIDENCE",
                    "is_retryable": False, "context": f"Confidence {confidence} below 0.5 threshold"}
        total_filings = sum(c.get("filing_count", 0) for c in merge_candidates) + primary_entity.get("filing_count", 0)
        total_lien_exposure = sum(c.get("estimated_amount", 0) for c in merge_candidates) + primary_entity.get("estimated_amount", 0)
        return {"is_error": False,
                "merged_profile_id": f"MP-{primary_entity.get('name', 'unknown')[:10]}-{confidence:.0%}",
                "canonical_name": primary_entity.get("name", "Unknown"),
                "total_filings": total_filings,
                "total_lien_exposure": total_lien_exposure,
                "states": list(set([primary_entity.get("state", "")] + [c.get("state", "") for c in merge_candidates])),
                "confidence": confidence,
                "merge_log": [f"Merged '{c.get('name', '')}' (score: {c.get('match_score', 'N/A')})" for c in merge_candidates]}
    except Exception as e:
        return {"is_error": True, "error_category": "INTERNAL_ERROR", "is_retryable": True, "context": str(e)}

python -c "from entity_tools import search_filings_by_name; print(search_filings_by_name('Acme Logistics LLC'))"

# entity_agent.py — ReAct Entity Resolution Agent
import anthropic, json
from entity_tools import (search_filings_by_name, fuzzy_match_score,
    get_filing_details, get_business_registry_data, merge_entity_profile)

SYSTEM_PROMPT = """You are an Entity Resolution Agent for a commercial data provider.
Given a business entity name, you determine whether UCC filing records
under different name variations refer to the same real-world company.

## Your Reasoning Process (ReAct)
For each resolution request, think step by step:
1. THINK: What do I know? What do I need to find out?
2. ACT: Call the appropriate tool to gather evidence.
3. OBSERVE: What did the tool return? What does it tell me?
4. REPEAT until you have enough evidence to make a decision.

## Decision Criteria
- token_sort_ratio >= 0.90 AND same state AND same address → MERGE (high confidence)
- token_sort_ratio >= 0.80 AND same state → LIKELY MERGE (verify with registry)
- token_sort_ratio >= 0.70 AND different state → INVESTIGATE (check registry)
- token_sort_ratio < 0.70 → DISTINCT ENTITY

## Output Format
After resolution, call merge_entity_profile with your findings.
Include a confidence score (0.0-1.0) based on evidence strength.
Explain your reasoning for each merge/separate decision.

## Rules
- Always check the business registry for ambiguous matches.
- If registry data is unavailable, lower your confidence accordingly.
- Never force a merge — flag conflicts for human review.
- If there are 10+ candidates, filter by state before matching."""

TOOLS = [
    {"name": "search_filings_by_name", "description": "Search UCC filings by business name across states. Returns candidate entities with filing counts.",
     "input_schema": {"type": "object", "properties": {"business_name": {"type": "string"}, "state": {"type": "string"}, "match_type": {"type": "string", "enum": ["exact", "fuzzy"]}}, "required": ["business_name"]}},
    {"name": "fuzzy_match_score", "description": "Compute similarity scores between two entity names using multiple algorithms.",
     "input_schema": {"type": "object", "properties": {"entity_a": {"type": "string"}, "entity_b": {"type": "string"}}, "required": ["entity_a", "entity_b"]}},
    {"name": "get_filing_details", "description": "Get full filing details for an entity in a state.",
     "input_schema": {"type": "object", "properties": {"business_name": {"type": "string"}, "state": {"type": "string"}}, "required": ["business_name", "state"]}},
    {"name": "get_business_registry_data", "description": "Cross-reference entity against official SOS business registry.",
     "input_schema": {"type": "object", "properties": {"business_name": {"type": "string"}, "state": {"type": "string"}}, "required": ["business_name", "state"]}},
    {"name": "merge_entity_profile", "description": "Create a merged entity profile from confirmed matches.",
     "input_schema": {"type": "object", "properties": {"primary_entity": {"type": "object"}, "merge_candidates": {"type": "array"}, "confidence": {"type": "number"}}, "required": ["primary_entity", "merge_candidates", "confidence"]}},
]

HANDLERS = {
    "search_filings_by_name": lambda a: search_filings_by_name(a["business_name"], a.get("state"), a.get("match_type", "fuzzy")),
    "fuzzy_match_score": lambda a: fuzzy_match_score(a["entity_a"], a["entity_b"]),
    "get_filing_details": lambda a: get_filing_details(a["business_name"], a["state"]),
    "get_business_registry_data": lambda a: get_business_registry_data(a["business_name"], a["state"]),
    "merge_entity_profile": lambda a: merge_entity_profile(a["primary_entity"], a["merge_candidates"], a["confidence"]),
}

def run_entity_agent(query: str) -> str:
    client = anthropic.Anthropic()  # reads ANTHROPIC_API_KEY env var
    messages = [{"role": "user", "content": query}]

    for _ in range(15):  # entity resolution may need many tool calls
        try:
            response = client.messages.create(
                model="claude-sonnet-4-6", max_tokens=4096,
                system=SYSTEM_PROMPT, tools=TOOLS, messages=messages,
            )
        except anthropic.APIError as e:
            return f"Error: {e}"

        if response.stop_reason == "tool_use":
            results = []
            for block in response.content:
                if block.type == "tool_use":
                    handler = HANDLERS.get(block.name)
                    result = handler(block.input) if handler else {"is_error": True, "error_category": "UNKNOWN_TOOL"}
                    results.append({"type": "tool_result", "tool_use_id": block.id, "content": json.dumps(result)})
            messages.append({"role": "assistant", "content": response.content})
            messages.append({"role": "user", "content": results})
        elif response.stop_reason == "end_turn":
            return " ".join(b.text for b in response.content if hasattr(b, "text"))

    return "Resolution exceeded maximum iterations."

if __name__ == "__main__":
    result = run_entity_agent("Resolve entity: Acme Logistics LLC in Delaware. Check for name variations across states.")
    print(result)

python entity_agent.py

# e2e_test.py — Quick verification script
from entity_tools import (search_filings_by_name, fuzzy_match_score,
    get_filing_details, get_business_registry_data, merge_entity_profile)

# 1. Search works
result = search_filings_by_name("Acme Logistics LLC")
assert not result["is_error"], "Search failed"
assert result["total"] == 3, f"Expected 3 candidates, got {result['total']}"
print("Search:   OK — 3 candidates found")

# 2. Fuzzy matching works
score = fuzzy_match_score("Acme Logistics LLC", "ACME LOGISTICS, L.L.C.")
assert not score["is_error"], "Fuzzy match failed"
assert score["recommendation"] == "likely_match", f"Expected likely_match, got {score['recommendation']}"
print(f"Fuzzy:    OK — {score['recommendation']} (token_sort: {score['scores']['token_sort_ratio']})")

# 3. Registry lookup works
reg = get_business_registry_data("Acme Logistics LLC", "DE")
assert not reg["is_error"], "Registry lookup failed"
print(f"Registry: OK — {reg['entity_name']} in {reg['state']}")

# 4. Merge works
merged = merge_entity_profile(
    {"name": "Acme Logistics LLC", "state": "DE", "filing_count": 3},
    [{"name": "ACME LOGISTICS, L.L.C.", "state": "DE", "match_score": 0.98, "filing_count": 1}],
    confidence=0.94
)
assert not merged["is_error"], "Merge failed"
print(f"Merge:    OK — {merged['canonical_name']}, confidence: {merged['confidence']}")

# 5. Error handling works
empty = search_filings_by_name("Nonexistent Corp XYZ")
assert empty["is_error"], "Expected error for unknown entity"
print(f"Errors:   OK — graceful handling for unknown entities")

print("\nAll checks passed. Run 'python entity_agent.py' for the full agent test.")