AI 工作流专栏 16：多步编排与人机协同 HITL

# sequence.py：顺序结构，最朴素的"流水线"
def classify_ticket(text: str) -> str:
    """第 1 步：分类工单"""
    # 实际项目里这里是 LLM 调用，先用假数据演示结构
    if "退款" in text:
        return "退款类"
    return "咨询类"

def route(category: str) -> str:
    """第 2 步：根据分类决定走哪条路（这里先返回处理方式）"""
    return f"按【{category}】流程处理"

def archive(result: str) -> None:
    """第 3 步：归档"""
    print(f"已归档：{result}")

# 顺序执行：一步的输出 = 下一步的输入
ticket = "我要退款，订单号 12345"
step1 = classify_ticket(ticket)   # 1. 分类
step2 = route(step1)              # 2. 路由（依赖 step1）
archive(step2)                    # 3. 归档（依赖 step2）

# branch.py：分支结构，根据分类走不同处理
def classify(ticket: str) -> str:
    if "退款" in ticket:
        return "refund"
    if "投诉" in ticket:
        return "complaint"
    return "faq"

def handle_refund(ticket: str):
    print("走退款流程：查订单 → 校验金额 → 触发审批")

def handle_complaint(ticket: str):
    print("走投诉流程：安抚话术 → 升级主管")

def handle_faq(ticket: str):
    print("走 FAQ 流程：直接 RAG 检索回答")

ticket = "这个商品坏了，我要退款"
category = classify(ticket)

# 多分支：根据分类结果选不同函数
if category == "refund":
    handle_refund(ticket)
elif category == "complaint":
    handle_complaint(ticket)
else:
    handle_faq(ticket)

# loop.py：循环结构，重试与迭代优化
import os, time
from openai import OpenAI

client = OpenAI(api_key=os.getenv("OPENAI_API_KEY"))

# --- 模式 A：重试（失败就再来，最多 N 次）---
def call_llm_with_retry(prompt: str, max_retries: int = 3):
    for attempt in range(1, max_retries + 1):
        try:
            resp = client.chat.completions.create(
                model="gpt-4o-mini",
                messages=[{"role": "user", "content": prompt}],
            )
            return resp.choices[0].message.content
        except Exception as e:
            print(f"第 {attempt} 次失败：{e}")
            if attempt == max_retries:
                raise  # 重试次数用尽，往上抛
            time.sleep(2 ** attempt)  # 指数退避：2s, 4s, 8s...

# --- 模式 B：迭代优化（结果不达标就改进，直到达标或超次数）---
def generate_until_good(question: str, max_rounds: int = 3):
    draft = call_llm_with_retry(f"回答这个问题：{question}")
    for round_ in range(1, max_rounds + 1):
        # 用 LLM 当"评委"，给草稿打分
        score_resp = client.chat.completions.create(
            model="gpt-4o-mini",
            messages=[{"role": "user", "content":
                f"给下面回答打 1-10 分，只输出数字。问题：{question}\n回答：{draft}"}],
        )
        score = int(score_resp.choices[0].message.content.strip())
        print(f"第 {round_} 轮，得分 {score}")
        if score >= 8:
            return draft  # 达标，退出循环
        # 不达标，让 LLM 改进一版
        draft = call_llm_with_retry(
            f"这个回答得分 {score}，请改得更好。问题：{question}\n原回答：{draft}")
    return draft  # 改到上限还不行，返回最后一版

print(generate_until_good("用大白话解释什么是 RAG"))

# parallel.py：并行结构，用线程池同时跑多个独立任务
import os, time
from concurrent.futures import ThreadPoolExecutor, as_completed
from openai import OpenAI

client = OpenAI(api_key=os.getenv("OPENAI_API_KEY"))

def ask(question: str) -> str:
    """一个独立的 LLM 调用任务"""
    resp = client.chat.completions.create(
        model="gpt-4o-mini",
        messages=[{"role": "user", "content": question}],
    )
    return resp.choices[0].message.content

questions = [
    "用一句话解释 RAG",
    "用一句话解释 Agent",
    "用一句话解释 Embedding",
    "用一句话解释向量数据库",
]

# --- 串行：一个一个问，总耗时约等于单次乘以 4 ---
t0 = time.time()
serial = [ask(q) for q in questions]
print(f"串行耗时 {time.time()-t0:.1f}s")

# --- 并行：4 个问题同时问，总耗时约等于最慢那一次 ---
t0 = time.time()
with ThreadPoolExecutor(max_workers=4) as pool:
    futures = {pool.submit(ask, q): q for q in questions}
    parallel = [f.result() for f in as_completed(futures)]
print(f"并行耗时 {time.time()-t0:.1f}s")

# hitl_workflow.py：用状态机 + Redis 实现"暂停-恢复"
# pip install redis
import os, json, uuid, redis

r = redis.Redis(host=os.getenv("REDIS_HOST", "localhost"),
                port=int(os.getenv("REDIS_PORT", "6379")), decode_responses=True)

# 把一个工作流的执行状态存进 Redis（"存档"）
def save_state(run_id: str, state: dict):
    state["updated_at"] = __import__("time").time()
    r.set(f"wf:{run_id}", json.dumps(state, ensure_ascii=False))

# 读出来（"读档"）
def load_state(run_id: str) -> dict | None:
    data = r.get(f"wf:{run_id}")
    return json.loads(data) if data else None

# ---- 一个"退款审批"工作流：分类 → 校验 → 【HITL 等审批】 → 执行退款 ----
def classify_and_validate(ticket: str) -> dict:
    """第 1、2 步：分类 + 校验（省略 LLM 细节）"""
    amount = 500  # 假装从订单系统查到的金额
    return {"ticket": ticket, "amount": amount, "category": "退款"}

def run_until_human(run_id: str, ticket: str):
    """跑流水线，直到撞上 HITL 节点就暂停"""
    state = load_state(run_id) or {}
    step = state.get("step", "start")

    if step == "start":
        state = classify_and_validate(ticket)
        state["step"] = "validated"
        save_state(run_id, state)

    if state["step"] == "validated":
        # HITL 节点：暂停，等人审批
        state["step"] = "awaiting_approval"
        save_state(run_id, state)
        print(f"已暂停，等待审批。run_id={run_id}，金额={state['amount']}")
        print("  请调用 resume(run_id, 'approve' 或 'reject')")
        return state  # 主动退出，把控制权交还

    if state["step"] == "approved":
        do_refund(state)            # 真正执行退款
        state["step"] = "done"
        save_state(run_id, state)
        return state

def do_refund(state):
    print(f"已退款 {state['amount']} 元")

def resume(run_id: str, decision: str):
    """人工审核后，从断点恢复"""
    state = load_state(run_id)
    if not state:
        raise ValueError("存档不存在，可能已过期")
    if state["step"] != "awaiting_approval":
        raise ValueError(f"当前状态 {state['step']} 不在等待审批")
    if decision == "approve":
        state["step"] = "approved"
        save_state(run_id, state)
        return run_until_human(run_id, state["ticket"])  # 读档后接着跑
    else:
        state["step"] = "rejected"
        save_state(run_id, state)
        print("已拒绝，流程终止")
        return state

# ---- 跑一把 ----
if __name__ == "__main__":
    rid = str(uuid.uuid4())[:8]
    run_until_human(rid, "商品坏了，申请退款")  # 会停在 awaiting_approval
    print("--- 此时服务可以挂掉、可以重启，状态都在 Redis 里 ---")
    resume(rid, "approve")                          # 人工点同意，从断点恢复

state = {
    "run_id": "abc123",            # 唯一标识（号码牌）
    "step": "awaiting_approval",   # 当前在哪一步（状态机的"状态"）
    "ticket": "...",               # 业务输入
    "category": "退款",            # 中间结果
    "amount": 500,                 # 中间结果
    "history": [                   # 每步的执行记录（审计用）
        {"step": "classify", "ok": True, "at": 1735689600},
        {"step": "validate", "ok": True, "at": 1735689605},
    ],
    "error": None,                 # 出错信息
    "updated_at": 1735689610,      # 最后更新时间
}

# error_handling.py：错误处理四板斧
import logging
logger = logging.getLogger("wf")

def with_retry(fn, max_retries=3, backoff=2):
    """板斧 1：重试 + 指数退避"""
    import time
    for i in range(1, max_retries + 1):
        try:
            return fn()
        except Exception as e:
            if i == max_retries:
                raise
            logger.warning(f"第 {i} 次失败：{e}，{backoff**i}s 后重试")
            time.sleep(backoff ** i)

def step_optional(fn):
    """板斧 2：跳过（失败不影响主线）"""
    try:
        return fn()
    except Exception as e:
        logger.warning(f"非关键步骤失败，跳过：{e}")
        return None

def step_critical(fn, on_fail):
    """板斧 3：关键步骤失败，降级到人工"""
    try:
        return fn()
    except Exception as e:
        logger.error(f"关键步骤失败，转人工：{e}")
        return on_fail(e)  # on_fail 通常是把工单推进人工队列

# 板斧 4 回滚通常需要结合事务，伪代码示意：
def refund_with_rollback(order_id):
    saved = backup_state(order_id)   # 先备份
    try:
        deduct_stock(order_id)
        do_payment(order_id)
    except Exception:
        restore_state(order_id, saved)  # 失败则回滚
        raise

# customer_support_pipeline.py：智能客服工单流水线（骨架版）
import os, json, uuid, logging
import redis
from openai import OpenAI

logger = logging.getLogger("support")
r = redis.Redis(host=os.getenv("REDIS_HOST","localhost"),
                port=int(os.getenv("REDIS_PORT","6379")), decode_responses=True)
client = OpenAI(api_key=os.getenv("OPENAI_API_KEY"))

# ---------- 状态存取 ----------
def save(rid, state): r.set(f"ticket:{rid}", json.dumps(state, ensure_ascii=False))
def load(rid):
    d = r.get(f"ticket:{rid}"); return json.loads(d) if d else None

# ---------- 单步函数 ----------
def classify(ticket: str) -> tuple[str, float]:
    """步骤1：LLM 分类，返回 (类别, 置信度)"""
    resp = client.chat.completions.create(
        model="gpt-4o-mini",
        messages=[{"role":"user","content":
          f"判断工单类别，只输出 咨询/投诉/退款 和置信度(0-1)。工单：{ticket}"}],
    )
    # 真实项目用 structured output 解析，这里简化
    return "退款", 0.92   # 假装解析出来

def rag_answer(ticket: str) -> tuple[str, float]:
    """步骤：RAG 检索 + 生成，返回 (答案, 置信度)"""
    # 检索细节此处省略
    return "您好，关于这个问题……", 0.6

def improve(answer: str, ticket: str) -> str:
    """循环里：改写答案"""
    resp = client.chat.completions.create(
        model="gpt-4o-mini",
        messages=[{"role":"user","content":f"改进这个回答。工单：{ticket}\n回答：{answer}"}])
    return resp.choices[0].message.content

def do_refund(rid, amount):
    logger.info(f"run={rid} 执行退款 {amount}")

# ---------- 主流程 ----------
def run(rid: str, ticket: str):
    state = load(rid) or {"ticket": ticket}
    step = state.get("step", "classify")

    # 步骤1：分类
    if step == "classify":
        category, conf = classify(ticket)
        state.update(category=category, classify_conf=conf)
        state["step"] = "routed"
        save(rid, state)

    # 步骤2：分支
    if state["step"] == "routed":
        cat = state["category"]
        if cat == "咨询":
            state["step"] = "answer"; save(rid, state)
        elif cat == "投诉":
            state["step"] = "escalate"; save(rid, state)   # 转 HITL 接管
        elif cat == "退款":
            state["step"] = "refund_check"; save(rid, state)

    # 步骤3：咨询类，RAG + 循环优化
    if state["step"] == "answer":
        answer, score = rag_answer(ticket)
        for round_ in range(1, 4):                       # 最多 3 轮
            if score >= 0.8:
                break
            answer = improve(answer, ticket)
            # 重新打分（省略）
            score = min(score + 0.1, 0.85)
        state.update(answer=answer, answer_conf=score)
        # 步骤4：置信度低转 HITL 人工
        state["step"] = "approve_answer" if score < 0.7 else "done_low_conf"
        save(rid, state)

    # 步骤4a：答案 HITL 审批
    if state["step"] == "approve_answer":
        state["step"] = "awaiting_answer_approval"
        save(rid, state)
        print(f"答案待审批，run={rid}")
        return  # 暂停

    # 步骤3b：退款类，HITL 审批
    if state["step"] == "refund_check":
        state["amount"] = 500                            # 假装查订单
        state["step"] = "awaiting_refund_approval"
        save(rid, state)
        print(f"退款待审批，run={rid}，金额={state['amount']}")
        return  # 暂停

    # 步骤3c：投诉，直接 HITL 接管
    if state["step"] == "escalate":
        state["step"] = "awaiting_human_takeover"
        save(rid, state)
        print(f"投诉转人工接管，run={rid}")
        return

    # 终态：归档
    if state["step"].startswith("done"):
        archive(rid, state)
        return

def archive(rid, state):
    logger.info(f"归档 run={rid} 类别={state.get('category')}")
    r.delete(f"ticket:{rid}")   # Redis 删掉，真实项目再写一份进 Postgres

# ---------- 恢复入口（人工审批后调用）----------
def resume(rid: str, decision: str, edited_answer: str = None):
    state = load(rid)
    if not state: raise ValueError("存档不存在")
    s = state["step"]
    if s == "awaiting_answer_approval":
        if decision == "approve":
            state["step"] = "done_approved"
        else:  # reject 或 edit
            state["answer"] = edited_answer or state["answer"]
            state["step"] = "done_approved"
        save(rid, state); run(rid, state["ticket"])
    elif s == "awaiting_refund_approval":
        if decision == "approve":
            do_refund(rid, state["amount"])
            state["step"] = "done_refunded"; save(rid, state); run(rid, state["ticket"])
        else:
            state["step"] = "done_rejected"; save(rid, state)
    elif s == "awaiting_human_takeover":
        state["step"] = "done_human"; save(rid, state)

if __name__ == "__main__":
    rid = str(uuid.uuid4())[:8]
    run(rid, "我买的耳机坏了，要退款")
    # 会停在 awaiting_refund_approval
    # （此时服务可重启，状态在 Redis）
    resume(rid, "approve")   # 主管点同意，自动退款 + 归档