智能体式 AI 驱动的支付编排

# Load config from MongoDB
config = db.conversion_configs.find_one({"_id": "MT103_to_JSON"})
# Extract fields using patterns from config["extract"]
for field_id, pattern in config["extract"].items():
    extracted[field_id] = re.search(pattern, message).group(1)
# Transform using rules from config["map"]
for mapping in config["map"]:
    value = extracted[mapping["from"]]
    if "ai" not in mapping:                  # RULES lane
        output[mapping["to"]] = value
    else:                                    # AI lane
        ai_fields.append({"value": value, "field_type": mapping["ai"]})

# Aggregation pipeline: extract field mappings from ALL existing configs
pipeline = [
    {"$addFields": {"extract_array": {"$objectToArray": "$extract"}}},
    {"$unwind": "$map"},
    {"$project": {"field_id": "$map.from", "mapping": "$map", "source_config": "$_id"}},
    {"$group": {"_id": "$field_id", "mapping": {"$first": "$mapping"}}}
]
field_lookup = {doc["_id"]: doc for doc in db.conversion_configs.aggregate(pipeline)}
# Match detected fields against learned patterns
for field_id in detected_fields:
    if field_id in field_lookup:
        matched[field_id] = field_lookup[field_id]["mapping"]
    else:
        unknown.append(field_id)
# For unknown fields: LLM suggests mappings constrained by format spec
suggestions = bedrock.invoke_claude(f"Map {unknown} to {target_format} fields")

from langchain_core.tools import tool
from langgraph.prebuilt import create_react_agent
@tool
def atlas_search(collection: str, query: str, search_fields: list, fuzzy: bool = True) -> dict:
    """Search MongoDB using Atlas Search. Use fuzzy=False for exact match first."""
    pipeline = [{"$search": {
        "index": f"{collection}_search",
        "text": {"query": query, "path": search_fields, "fuzzy": {"maxEdits": 2} if fuzzy else {}}
    }}]
    results = list(db[collection].aggregate(pipeline))
    return {"found": bool(results), "top_result": results[0] if results else None}
@tool
def vector_search(collection: str, query: str) -> dict:
    """Semantic search using Atlas Vector Search - matches by meaning, not keywords."""
    embedding = voyage_client.embed([query], model="voyage-3").embeddings[0]
    pipeline = [{"$vectorSearch": {
        "index": f"{collection}_vector", "path": "embedding",
        "queryVector": embedding, "numCandidates": 100, "limit": 3
    }}]
    results = list(db[collection].aggregate(pipeline))
    return {"found": bool(results), "top_result": results[0] if results else None}
agent = create_react_agent(model=llm, tools=[atlas_search, vector_search])

# Step 1: Convert incoming message to canonical JSON
hop1_result = await converter.convert(
    source_format="pacs.008",  # or MT103, ISO8583, etc.
    target_format="JSON",
    message=raw_message
)
# Step 2: Parse the canonical JSON from conversion result
canonical_json = json.loads(hop1_result['converted_message'])
# Step 3: Check canonical JSON for crypto settlement fields
if canonical_json.get('crypto_blockchain') and canonical_json.get('crypto_receiver_wallet'):
    # Route to Solana
    solana_service.transfer(
        to_pubkey=canonical_json['crypto_receiver_wallet'],
        amount_sol=50000
    )
# step 4: Solana Service Class
class SolanaService:
    def transfer(self, to_pubkey: str, amount_sol: float) -> dict:
        response = self.client.send_raw_transaction(bytes(tx))
        self.client.confirm_transaction(response.value)
        return {
            "signature": str(response.value),
            "explorer_url": f"https://explorer.solana.com/tx/{response.value}?cluster=devnet"
        }