手搓一个催收Agent（三）：架构设计与伪代码

[geoqiao]

一句话结论:催收Agent就是一个按照固定流程执行任务的Agent:查看用户→检查合规→触达→识别意图→加载策略→调用工具。四个组件(Context/Permissions/Decider/Skills/Tools)各负责一个环节。

一、流程总览

flowchart LR
    A[Context<br/>查看用户列表] --> B[Permissions<br/>检查合规]
    B --> C[Decider<br/>识别意图]
    C --> D[Skill<br/>选择并执行应对策略]
    D --> E[Tool<br/>策略中的具体动作]

Loading

催收Agent的核心流程只有四步：

流程环节	Agent组件	职责
查看逾期用户列表	Context	加载用户画像、历史对话、系统数据
检查能否触达	Permissions	检查时间、频次、DNC、敏感职业
识别用户意图	Decider	根据上下文判断用户属于哪种意图
选择并执行应对策略	Skill + Tool	加载对应Skill策略，执行时调用Tool

关键理解：Skill 和 Tool 不是并列关系。Skill 是"策略模板"（协商/安抚/争议处理），Tool 是策略执行时的"具体动作"（查账单、发链接、记录承诺）。Agent 先根据意图选择 Skill，再在 Skill 的 ReAct 循环中决定调用哪些 Tool。

二、Context:查看用户列表

Agent的第一步:加载所有需要的信息。

class UserState:
    user_id: str
    profile: UserProfile      # 金额、逾期天数、职业
    conversation: ConversationContext  # 最近50条对话
    intent_history: list[str]  # 历史意图
    dnc: bool                  # 是否要求停止

class Context:
    user_message: str | None
    facts: dict                # 系统注入的事实
    messages: list[Message]    # 对话历史
    session_state: str         # idle/interacting/stopped

def build_context(user_id):
    state = db.load(user_id)
    return Context(
        facts={
            "due_amount": state.profile.amount_due,
            "overdue_days": state.profile.overdue_days,
        },
        messages=state.conversation.messages[-50:],
    )

关键设计:金额和逾期天数从系统读取,作为facts注入,LLM不能编造。

三、Permissions:检查合规

Agent的第二步:在触达前检查硬规则。

class Harness:
    def check(self, event, state) -> HarnessResult:
        # 状态拦截
        if state.dnc or state.session_state == "stopped":
            return block("user_stopped")

        # 时间窗口
        if not compliance.is_within_valid_hours():
            return block("outside_valid_hours")

        # 频次限制
        if quota.exceeded(state.user_id):
            return block("quota_exceeded")

        # STOP/CRISIS关键词快速路径
        if stop_keyword in event.message:
            return force_intent("STOP")

        # 敏感职业标记
        if state.profile.is_sensitive:
            return mark_sensitive()

关键设计:所有检查在LLM调用前完成。不通过直接返回,不进后续流程。

四、Decider：触达+识别意图

Agent的第三步：执行触达，识别用户意图。

意图来源不只是用户说了什么

在催收场景中，意图识别的输入不局限于用户的文本回复。以下都是意图识别的来源：

输入类型	示例	对应的意图
用户文本	"我现在没钱"	NEGOTIATION
用户不输入	电话未接、短信不回	UNREACHABLE
用户行为	点击了还款链接但没支付	AVOIDANCE
用户信息	职业=律师（敏感职业）	强制标准话术
用户状态	已还清、已加入DNC	STOP
系统事件	承诺还款时间到期未还	NEGOTIATION（违约跟进）

这意味着 Decider 的输入是整个 Context，而不仅是 user_message。

class Decider:
    async def decide(self, context: Context) -> Decision:
        prompt = f"""
        用户消息：{context.user_message or "（无输入）"}
        系统facts：{context.facts}
        历史意图：{context.intent_history}
        用户状态：{context.session_state}
        触达结果：{context.last_outreach_result}

        基于以上全部信息识别用户意图，从以下8种中选择：
        COOPERATION, NEGOTIATION, AVOIDANCE, DISPUTE,
        STOP, CRISIS, REFUSE, UNREACHABLE

        注意：
        - 用户未回复（last_outreach_result=UNREACHABLE）→ UNREACHABLE
        - 承诺到期未还（overdue_promise=True）→ NEGOTIATION（违约跟进）
        - 用户状态=stopped → STOP

        输出JSON：{{"intent": "...", "confidence": 0.95}}
        """
        response = await llm.chat(prompt)
        return parse_intent(response)

意图映射到Skill:

意图	Skill
COOPERATION	payment_guidance
NEGOTIATION	negotiation
AVOIDANCE	reengage
DISPUTE	dispute
STOP	stop
CRISIS	crisis
REFUSE	troubleshoot
UNKNOWN	troubleshoot

五、Skill + Tool：选择并执行应对策略

Agent的第四步：根据意图选择 Skill 策略，在策略执行中调用 Tool。

意图 → Skill 映射

识别出意图后，直接映射到对应的 Skill：

意图	Skill	说明
COOPERATION	payment_guidance	引导还款
NEGOTIATION	negotiation	协商方案
AVOIDANCE	reengage	沉默追踪
DISPUTE	dispute	核实账单
STOP	stop	确认退出
CRISIS	crisis	危机响应
REFUSE	troubleshoot	兜底处理

这是规则映射，不需要 LLM。IntentClassifier 输出意图，SkillSelector 查表即可。

Skill：策略模板

Skill 是一个独立的策略文件，定义了"面对这种意图时，应该怎么处理"。

name: negotiation
description: 用户表示困难，希望延期或分期
tools: [query_bill, record_promise, send_payment_link]
max_steps: 5

你是协商专家。用户表示还款困难，你的目标是：

1. 了解困难原因
2. 提供可行方案（延期7天/14天、分3期）
3. 记录用户的还款承诺

禁止：施压、威胁、承诺超出权限的减免

Skill 的关键设计：

• tools：白名单，这个 Skill 只能调用这些 Tool
• max_steps：防止无限循环
• content：策略指令，告诉 LLM 应该怎么谈判

Tool：策略中的具体动作

Tool 是 Skill 执行时的"手脚"。Skill 的 ReAct 循环中，LLM 决定什么时候调用什么 Tool。

@tool("query_bill", "查询用户账单")
async def query_bill(user_id: str, store) -> dict:
    state = store.load(user_id)
    return {
        "amount_due": state.profile.amount_due,
        "overdue_days": state.profile.overdue_days,
    }

@tool("send_payment_link", "发送还款链接")
async def send_payment_link(user_id: str, amount: float) -> dict:
    link = generate_link(user_id, amount)
    await send_sms(user_id, f"还款链接：{link}")
    return {"sent": True, "link": link}

@tool("record_promise", "记录还款承诺")
async def record_promise(user_id: str, amount: float, date: str, store) -> dict:
    store.save_promise(user_id, amount, date)
    return {"recorded": True}

ReAct 循环：Skill 执行时调用 Tool

class SkillExecutor:
    async def execute(self, skill, context, state) -> SkillResult:
        messages = build_messages(skill, context)

        for step in range(skill.max_steps):
            response = await llm.chat(messages)
            action = parse_action(response)

            if action.type == "reply":
                return SkillResult(response_text=action.text)

            elif action.type == "tool_call":
                # 只能调用Skill白名单中的工具
                if action.name not in skill.tools:
                    return error("tool_not_allowed")

                result = await call_tool(action.name, action.params, state)
                messages.append(f"Tool result: {result}")

        return error("max_steps_exceeded")

一个完整的 ReAct 示例：

Step 1: LLM 思考 → "用户说没钱，我需要先查账单"
Step 1: LLM 调用 tool → query_bill(user_id="U-8842")
Step 1: Tool 返回 → {"amount_due": 2580, "overdue_days": 15}

Step 2: LLM 思考 → "金额2580，逾期15天。用户说没钱，我提供分期方案"
Step 2: LLM 回复 → "理解您的困难。您的账单是2580元，可以分3期..."

Tool 分级

分级	工具	风险	说明
只读	query_bill, query_user_history	无	查询信息
扩展	generate_payment_link, send_payment_link	低	发送内容
写入	record_promise, escalate_to_cs, add_to_dnc	高	修改状态

六、完整串联:AgentSession

class AgentSession:
    async def handle_event(self, event: Event) -> SkillResult:
        # 1. 构建Context(查看用户列表)
        context = build_context(event.user_id)

        # 2. Permissions检查(选择触达方式前的合规检查)
        harness_result = self.harness.check(event, self.user_state)
        if harness_result.block:
            return blocked_result(harness_result.reason)

        # 3. Decider：识别意图
        decision = await self.decider.decide(context)

        # 4. 根据意图选择 Skill
        skill = self.skill_registry.get(decision.selected_skill)

        # 5. 执行 Skill（ReAct循环，内部调用Tools）
        result = await self.skill_executor.execute(
            skill, context, self.storage
        )

        # 6. 输出审核
        if not self.compliance.audit_content(result.response_text):
            result.response_text = fallback_message()

        # 7. 保存状态
        self.save_state()
        return result

关联

• [[02-collector-workflow]] - 上一篇:催收员的真实工作流程
• [[chatbot-exploration]] - 催收chatbot项目设计文档

Tags

#collection-agent #ai-engineering #agent #architecture