《实用笔记》：构建可用于生产的智能欺诈检测系统

本指南将逐步构建从原材料到可运行系统的完整流程，内容为：打造可用于生产的智能欺诈检测系统——第一部分：整体概览。重点在于可操作的步骤、明确的检查项，以及可直接放入代码仓库的代码，无需猜测其用途。 在概览阶段，应在修改代码之前明确输入参数、各步骤的负责人以及完成标准。操作人员应能够从已知的检查点重新运行相应步骤，而无需猜测隐藏的状态。 可将此阶段视为输入与经过验证的输出之间的契约。为相关成果命名，定义成功检查标准，并拒绝默许的半完成状态。

问题的形态

在处理“舞台的形态”这一任务时，首先列出相关约定：所需的输入参数、成功信号以及部分失败时的处理方式。这样的检查清单能确保后续的代码修改保持一致性。 在功能结果旁记录执行时间以及令牌或查询成本。提前明确成本信息，可避免在从演示环境过渡到共享环境时出现意外费用。 在耗时较高的步骤之后设置检查点。当操作员重新执行后续节点时，恢复流程不应再次收取相同的LLM调用费用。

系统概览

在完成系统概览阶段时，首先列出相关约定：所需输入、成功信号以及部分失败时的处理方式。这样的清单能确保后续的代码修改保持一致性。 将配置信息与应用程序代码分开存放。环境文件、密钥存储以及功能开关应集中于一个位置，这样操作人员无需查看整个系统结构即可进行审计。 在耗时较高的步骤之后设置检查点。当操作人员重新执行后续节点时，恢复流程不应再次调用相同的大型语言模型。

智能编排——实时欺诈评分

在处理“代理编排实时”阶段时，首先需明确合同条款：所需输入、成功信号以及部分失败时的处理方式。这样的检查清单能确保后续的代码修改保持一致性。 同时记录正常流程与恢复流程。重试机制、人工审核环节以及死信处理都是产品本身的组成部分，而非后续的优化内容。 在成本较高的步骤之后设置检查点。当操作员重新尝试某个后续节点时，恢复流程不应再次调用相同的大型语言模型。

Agent A — 概率评分（训练好的模型）

在处理 Agent A 的倾向性评分阶段时，首先写下相关规范：所需输入、成功标志以及部分失败时的处理方式。这样的检查清单能确保后续的代码修改保持一致性。 优先选择小型、可测试的单元，而非庞大的脚本。当某个步骤失败时，故障应能指向单一责任模块，而非复杂的流程链。 缓存稳定的系统指令和工具结构。重复发送相同的开头信息是造成资源浪费的常见原因。

Agent B — 行为评分（完全不使用模型）

在处理 Agent B 的行为评分阶段时，首先写下相关契约：所需的输入参数、成功标志，以及部分失败时的处理方式。这样的检查清单能确保后续的代码修改保持一致性。 将这一阶段视为输入与验证后输出之间的契约。为相关成果命名，明确成功判定标准，杜绝无声的半完成状态。 缓存稳定的系统指令和工具架构。重复发送相同的开头信息是导致资源浪费的常见原因。

Agent C — 基于 pgvector 的策略检索

在处理 Agent C 的策略检索阶段时，首先记下相关要求：所需的输入参数、成功信号以及部分失败时的处理方式。这样的清单能确保后续的代码修改不会偏离原有设计。 在功能结果旁记录执行时间以及令牌或查询成本。提前了解成本情况，可避免在从演示环境过渡到共享环境时出现意外费用。 在调整提示词之前，先使用固定的问题集来测试召回率。仅仅更换提示词往往无法改善较差的检索效果。

工程化手段：模型卡片中未提及的部分

在处理 Harness 的工程设计及组件阶段时，首先需明确相关约定：所需的输入参数、成功信号以及部分故障时的处理方式。这样的检查清单能确保后续的代码修改始终符合预期。 应将配置信息与应用程序代码分开。环境文件、密钥存储以及功能开关应集中存放于一个位置，以便操作人员无需查看整个系统结构即可进行审计。 需缓存系统中稳定的指令及工具架构。重复发送相同的开头信息是导致资源浪费的常见原因。

"""
SessionMemory — the public interface tying working memory (short-term) and
episodic memory (long-term, cross-session) together for context engineering.

    mem = SessionMemory()                      # new session, fresh UUID
    mem.remember("user", "...")                # auto-persists to pgvector
    mem.remember("assistant", "...")
    context = mem.build_context(next_question) # summary + recent turns + relevant past episodes
    mem.end_session()                          # finalize into episodic_memory
"""

"""
In-memory semantic cache in front of episodic recall.

Two layers:
  - exact-string embedding cache: a literal repeat query skips the OpenAI
    embeddings API call entirely.
  - semantic result cache: a near-duplicate query (different wording, same
    intent) still needs embedding to compare, but skips the Postgres/pgvector
    round-trip if it's cosine-similar enough to something already cached.

Process-local only (not shared across workers/processes) — fine for a
single running service, not a substitute for a distributed cache if this
ever runs behind multiple instances.

Must be invalidated when a new episode is saved: a cached "no good match" or
partial result set can go stale the moment the underlying corpus changes.
"""

"""
Fraud-detection event chain, built on the generic pub-sub bus in
common/pubsub.py:

    UserQueryEvent
        -> InputGuardrailService   -> InputGuardrailPassedEvent | InputGuardrailBlockedEvent
        -> OrchestrationService    -> OrchestrationCompletedEvent | OrchestrationPausedHITLEvent
        -> OutputGuardrailService  -> OutputGuardrailPassedEvent | OutputGuardrailBlockedEvent
        -> ResultPublisher         -> PipelineCompletedEvent

    HumanDecisionEvent (resumes a run paused at OrchestrationPausedHITLEvent)
        -> OrchestrationService    -> ... (same chain onward)

Each service subscribes to exactly one (or two, for resume) event type and
publishes the next event in the chain — a new listener (audit logger,
LangSmith exporter) can subscribe to any event without touching the
publishers. The guardrail/graph calls underneath are synchronous
(psycopg2, HF/OpenAI SDKs); handlers run them via asyncio.to_thread so a
slow call doesn't block the event loop for other in-flight events.
"""

@mcp.tool()
@guarded(action="score_transaction")
def score_propensity(features: dict[str, float], role: str) -> float:
    """Score one transaction's V1-V28 features for ML fraud probability (0-1). Agent A.
    Requires role: analyst or admin."""
    return score_customer_propensity(features)

@mcp.tool()
@guarded(action="score_transaction")
def score_behavior(customer_id: int, amount: float, category: str, merchant: str, role: str) -> dict:
    """Score one transaction's behavior anomaly vs its category's peer-cohort stats. Agent B.
    Requires role: analyst or admin."""
    return score_customer_behavior(customer_id, amount, category, merchant)

@mcp.tool()
@guarded(action="chat_query", free_text_arg="query")
def consult_fraud_policy(query: str, role: str, k: int = 3) -> list[dict]:
    """Semantic search over the indexed credit-card policy/handbook documents. Agent C.
    Requires role: viewer or higher. `query` is scanned for prompt injection/jailbreak and PII."""
    return consult_policy(query, k=k)


if __name__ == "__main__":
    mcp.run()

JWT_SECRET_KEY = os.environ["JWT_SECRET_KEY"]
JWT_ALGORITHM = os.environ.get("JWT_ALGORITHM", "HS256")
JWT_EXPIRE_MINUTES = int(os.environ.get("JWT_EXPIRE_MINUTES", "30"))
pwd_context = CryptContext(schemes=["bcrypt"], deprecated="auto")
oauth2_scheme = OAuth2PasswordBearer(tokenUrl="token")

def verify_password(plain_password: str, hashed_password: str) -> bool:
    return pwd_context.verify(plain_password, hashed_password)

def get_password_hash(password: str) -> str:
    return pwd_context.hash(password)

def get_user(username: str) -> dict | None:
    conn = get_connection()
    try:
        with conn.cursor() as cur:
            cur.execute("SELECT * FROM users WHERE username = %s", (username,))
            return cur.fetchone()
    finally:
        conn.close()

def authenticate_user(username: str, password: str) -> dict | None:
    user = get_user(username)
    if not user or user["disabled"]:
        return None
    if not verify_password(password, user["hashed_password"]):
        return None
    return user

可观测性

在处理可观测性阶段时，首先需明确相关约定：所需的输入参数、成功标志，以及部分失败时的处理方式。这份清单能确保后续的代码修改保持一致性。 同时记录正常流程与故障恢复流程。重试机制、人工审核环节以及死信处理都是产品功能的一部分，而非后续需要补充的内容。 在成本较高的操作之后设置检查点。当操作员重新尝试某个节点时，恢复流程不应再次调用相同的大型语言模型接口。 在处理可观测性阶段时，首先需明确相关约定：所需的输入参数、成功标志，以及部分失败时的处理方式。这份清单能确保后续的代码修改保持一致性。 将此阶段视为输入与经过验证的输出之间的契约。为相关产物命名，定义成功检测标准，杜绝无声的半完成状态。

AWS部署与前端

将 AWS 部署与阶段视为可度量的对象，才能使其发挥最佳作用。在扩大范围之前，先记录一份理想的运行日志、一个故障案例以及回滚说明。在功能结果旁同时记录执行时间以及令牌或查询成本。提前了解成本情况，可避免在从演示环境过渡到共享环境时出现意外账单。要保持图表状态简洁且类型明确，嵌套的数据块会掩盖具体是哪个节点修改了哪个字段，还会在中断后导致无法继续处理。

下一步是什么

“下一步计划”阶段若被视为可度量的对象，效果会更好。在扩大范围之前，需记录一份最佳处理方案、一个故障案例以及回滚说明。

运营检查清单

“运营检查清单”阶段若被视为可度量的对象，效果会更好。在扩大范围之前，需记录一份最佳处理方案、一个故障案例以及回滚说明。

应优先选择小型且可测试的单元，而非庞大的脚本。当某一步骤失败时，故障应能指向具体的责任模块，而非复杂的流程链。

保持图结构的状态扁平且具有类型约束。嵌套的数据块会隐藏是哪个节点修改了哪个字段，还会在中断后导致流程无法继续。

只要预算允许，就在持续集成过程中使用测试用例而非真实的付费 API 来执行关键路径的冒烟测试。

将此阶段视为输入与经过验证的输出之间的契约。为相关产物命名，明确成功标准，绝不允许出现无声的半完成状态。

保持图结构的状态扁平且具有类型约束。嵌套的数据块会隐藏是哪个节点修改了哪个字段，还会在中断后导致流程无法继续。

在升级技术栈之前，先冻结版本，为关键路径生成标准化的操作记录，并确认回滚步骤。共享环境需要设置速率限制、进行租户身份验证，同时要明确负责密钥轮换的人员。与其追求华丽的临时演示，不如注重扎实的可靠性。

关于5d0d59733252的批量处理说明：不要将提供者密钥放入代码仓库，为每个会话设置令牌使用上限，并将转录内容存储在评估测试文件旁，以便后续更换模型时仍能保持可比性。

在处理强化安全措施的第0阶段时，首先明确需求规范：所需输入、成功标志以及部分失败时的处理方式。这样的检查清单能确保后续的代码修改始终符合要求。优先选择小型、可测试的单元，而非冗长的脚本；当某一步骤失败时，故障应能指向单一责任点，而非复杂的流程链。

强化安全措施细节0/721：需统计该处理步骤的运行时间、错误类型以及令牌消耗情况，然后依据固定的评估标准而非主观判断来决定是否保留该更改。

在将加固过程视为可测量的表面时，第一阶段的效果最佳。在扩大范围之前，先记录一份理想的运行结果、一个故障案例以及回滚说明。在功能结果旁还需记录执行时间以及令牌或查询成本。提前了解这些成本信息，就能避免在系统从演示环境过渡到共享环境时出现意外费用。

加固细节1/721：为该步骤测量实际执行时间、错误类型以及令牌消耗情况，然后依据固定的评估标准而非主观经验来决定是否保留该变更。

对于加固过程的第二阶段，在修改代码之前需明确输入参数、该步骤的负责人以及完成标准。操作人员应能够从已知的检查点重新运行该步骤，而无需猜测系统的隐藏状态。同时需将正常流程和故障恢复流程一并记录下来。重试机制、人工审核环节以及错误处理方式都是产品本身的一部分，而非后续需要补充的内容。

强化措施细节2/721：记录该代码段的运行时间、错误类型以及代币消耗情况，然后依据固定的评估标准而非个人经验来决定是否保留该修改。

在处理强化措施笔记的第三阶段时，首先需写出合约内容：所需的输入参数、成功标志，以及部分失败时的处理方式。这样的检查清单能确保后续的代码修改保持一致性。应将此阶段视为输入与验证后输出之间的契约，为相关组件命名、明确成功判定标准，并杜绝无声的半完成状态。

强化措施细节3/721：记录该代码段的运行时间、错误类型以及代币消耗情况，然后依据固定的评估标准而非个人经验来决定是否保留该修改。

将强化措施的第4阶段视为可测量的表面来处理效果最佳。在扩大范围之前，先记录一份理想的运行日志、一个故障案例以及回滚说明。 应将配置与应用程序代码分开。环境文件、密钥存储和功能标志应集中存放于一个位置，这样操作人员无需查看整个系统结构即可进行审计。

强化措施细节4/721：针对该措施需测量耗时、错误类型以及令牌使用情况，然后依据固定的问题集而非个人经验来判断是否保留该变更。