jaspersoft-agent-learn/step_02_state/exercise_answer.py

"""
Step 02 练习题答案

⚠️ 先自己思考，再看答案！
⚠️ 答案不是唯一的，这里只是其中一种实现
"""

from typing import TypedDict, List, Dict, Any
from datetime import datetime


# ═══════════════════════════════════════════════════════════════════════════════
# 练习 1 答案：客服聊天机器人状态
# ═══════════════════════════════════════════════════════════════════════════════

class CustomerServiceState(TypedDict, total=False):
    """客服聊天机器人的状态"""

    # === 用户信息 ===
    user_id: str                    # 用户 ID
    user_name: str                  # 用户名
    user_email: str                 # 用户邮箱（可选）

    # === 对话历史 ===
    conversation_history: List[dict]  # 对话历史
    """
    格式: [{"role": "user", "content": "..."},
          {"role": "assistant", "content": "..."}]
    """

    # === 问题处理 ===
    current_issue: str              # 当前正在处理的问题描述
    issue_status: str              # 问题状态: "open" / "investigating" / "resolved" / "closed"
    issue_priority: str             # 优先级: "low" / "medium" / "high" / "urgent"

    # === 解决方案 ===
    proposed_solution: str          # 提出的解决方案
    solution_steps: List[str]       # 解决步骤列表
    is_resolved: bool              # 是否已解决

    # === 用户反馈 ===
    satisfaction_rating: int        # 满意度评分 1-5
    feedback_comment: str           # 反馈意见

    # === 元信息 ===
    session_start: str              # 会话开始时间
    last_interaction: str           # 最后互动时间
    agent_id: str                   # 处理此会话的客服 ID


# ═══════════════════════════════════════════════════════════════════════════════
# 练习 2 答案：数据分析 Agent 状态
# ═══════════════════════════════════════════════════════════════════════════════

class DataAnalysisState(TypedDict, total=False):
    """数据分析 Agent 的状态"""

    # === 用户需求 ===
    user_request: str               # 用户的分析需求
    session_id: str                 # 会话 ID

    # === 数据源配置 ===
    data_source_type: str            # 数据源类型: "database" / "file" / "api"
    data_source_config: dict        # 数据源配置（连接信息等）
    """
    示例：
        {
            "host": "localhost",
            "database": "sales_db",
            "table": "orders"
        }
    """

    # === 查询和结果 ===
    query: str                      # 执行的 SQL 或查询条件
    query_result: Any               # 查询结果
    result_row_count: int           # 结果行数
    result_columns: List[str]       # 结果列名

    # === 分析过程 ===
    stage: str                      # 当前阶段
    """
    阶段值：
        - "initial": 初始状态
        - "connecting": 连接数据源
        - "querying": 执行查询
        - "analyzing": 分析数据
        - "generating_report": 生成报告
        - "completed": 完成
        - "error": 出错
    """

    # === 生成的报告 ===
    generated_report: str           # 生成的报告内容
    report_format: str              # 报告格式: "json" / "csv" / "markdown" / "html"

    # === 错误处理 ===
    error_message: str              # 错误信息（如果有）
    retry_count: int                # 重试次数


# ═══════════════════════════════════════════════════════════════════════════════
# 练习 3 答案：快照和恢复
# ═══════════════════════════════════════════════════════════════════════════════

def create_snapshot(state: dict) -> dict:
    """
    创建状态快照

    策略：只保存"业务关键"数据，不保存临时计算结果
    """
    # 定义需要保存的字段（业务关键数据）
    business_fields = [
        "user_id",
        "user_name",
        "current_task",
        "progress",
        "status",
        # 根据实际情况添加更多...
    ]

    snapshot_data = {}
    for field in business_fields:
        if field in state:
            snapshot_data[field] = state[field]

    return {
        "data": snapshot_data,
        "timestamp": datetime.now().isoformat(),
        "version": "1.0"
    }


def restore_from_snapshot(state: dict, snapshot: dict) -> dict:
    """
    从快照恢复状态
    """
    if not snapshot or "data" not in snapshot:
        return state

    # 从快照恢复数据
    snapshot_data = snapshot["data"]
    for key, value in snapshot_data.items():
        state[key] = value

    # 更新恢复后的时间戳
    state["_restored_at"] = datetime.now().isoformat()
    state["_restored_from"] = snapshot.get("timestamp", "unknown")

    return state


# ═══════════════════════════════════════════════════════════════════════════════
# 练习 4 答案：状态验证
# ═══════════════════════════════════════════════════════════════════════════════

def validate_state(state: dict, rules: dict) -> dict:
    """
    验证状态
    """
    errors = []

    for field_name, field_rules in rules.items():
        value = state.get(field_name)
        field_type = field_rules.get("type")
        required = field_rules.get("required", False)

        # 1. 检查必填
        if required and (value is None or value == ""):
            errors.append(f"字段 '{field_name}' 是必填的")
            continue

        # 如果字段为空且不是必填，跳过后续检查
        if value is None or value == "":
            continue

        # 2. 检查类型
        if field_type and not isinstance(value, field_type):
            errors.append(
                f"字段 '{field_name}' 类型错误: "
                f"期望 {field_type.__name__}, 实际 {type(value).__name__}"
            )
            continue

        # 3. 检查数值范围
        if isinstance(value, (int, float)):
            if "min" in field_rules and value < field_rules["min"]:
                errors.append(
                    f"字段 '{field_name}' 小于最小值: "
                    f"{value} < {field_rules['min']}"
                )
            if "max" in field_rules and value > field_rules["max"]:
                errors.append(
                    f"字段 '{field_name}' 大于最大值: "
                    f"{value} > {field_rules['max']}"
                )

        # 4. 检查枚举值
        if "choices" in field_rules:
            if value not in field_rules["choices"]:
                errors.append(
                    f"字段 '{field_name}' 值不在允许范围内: "
                    f"{value} not in {field_rules['choices']}"
                )

    return {
        "valid": len(errors) == 0,
        "errors": errors
    }


# ═══════════════════════════════════════════════════════════════════════════════
# 测试
# ═══════════════════════════════════════════════════════════════════════════════

def test_answers():
    """测试答案"""
    print("\n" + "=" * 60)
    print("测试练习答案")
    print("=" * 60)

    # 测试练习 1
    print("\n📝 练习 1: 客服状态")
    cs_state: CustomerServiceState = {
        "user_id": "user_001",
        "user_name": "张三",
        "issue_status": "open",
        "issue_priority": "high",
    }
    print(f"  状态: {cs_state}")

    # 测试练习 2
    print("\n📝 练习 2: 数据分析状态")
    da_state: DataAnalysisState = {
        "user_request": "分析本月销售数据",
        "data_source_type": "database",
        "stage": "querying",
    }
    print(f"  状态: {da_state}")

    # 测试练习 3
    print("\n📝 练习 3: 快照和恢复")
    sample_state = {
        "user_id": "123",
        "current_task": "数据分析",
        "progress": 50,
        "temp_data": ["计算中...", "处理中..."],
    }
    print(f"  原始状态: {sample_state}")

    snapshot = create_snapshot(sample_state)
    print(f"  快照: {snapshot}")

    sample_state["progress"] = 100
    restored = restore_from_snapshot(sample_state, snapshot)
    print(f"  恢复后状态: {restored}")

    # 测试练习 4
    print("\n📝 练习 4: 状态验证")
    rules = {
        "user_id": {"type": str, "required": True},
        "age": {"type": int, "min": 0, "max": 150},
        "status": {"type": str, "choices": ["active", "inactive"]}
    }

    valid_state = {"user_id": "123", "age": 25, "status": "active"}
    print(f"  有效状态: {validate_state(valid_state, rules)}")

    invalid_state = {"user_id": "123", "age": 200, "status": "unknown"}
    print(f"  无效状态: {validate_state(invalid_state, rules)}")


if __name__ == "__main__":
    test_answers()