第3章: 「灵语」智能对话中枢

# Round 1
用户：快递到哪了？
系统：我先帮你查物流。请提供订单号。

# Round 2
用户：订单 123，另外我想改地址。
系统：好的，我先返回物流状态，然后再处理改地址。请把新地址发我。

# Round 3：追问上下文
用户：我的订单号是多少
系统：您好，您之前提到的订单号是 123。如果您需要进一步处理退货或其他问题，请确认是否为该订单号，或提供新的订单号以便我为您查询和协助。

{
  "intent": ["track_shipping", "change_address"],  // 识别出的意图列表（支持多意图）
  "slots": {"order_id": "123", "new_address": "?"},  // 从输入中提取的结构化槽位（键值对）
  "confidence": 0.86  // 意图识别的置信度（0-1），低于阈值可转人工
}

{
  "trace_id": "9f1a...",        // 单次请求唯一标识，用于回放、定位问题
  "session_id": "abc...",      // 会话标识，关联同一用户的多轮对话
  "input": "订单123，另外我想改地址",  // 用户原始输入
  "intent": ["track_shipping", "change_address"],  // 识别出的意图列表（可多意图）
  "slots": {"order_id": "123", "new_address": "?"},  // 从输入中提取的结构化槽位
  "latency_ms": 820,           // 本次请求总耗时（毫秒），用于性能监控
  "prompt_version": "v1"       // Prompt 版本号，用于 A/B 测试与效果对比
}

# 最小"可组合模块"架构：Router + Memory + Tools + Prompt + Model
# 你只要记住一个关键：对话系统不是一个 prompt，而是一条可插拔流水线。

import os
from dataclasses import dataclass, field
from typing import Callable, Dict, List

from langchain_community.chat_models import ChatTongyi
from langchain_core.messages import AIMessage, BaseMessage, HumanMessage
from langchain_core.prompts import ChatPromptTemplate, MessagesPlaceholder
from langchain_core.runnables import RunnableLambda


# 🟢 记忆管理层（对应架构图第4层）：负责存储和裁剪对话历史
# 功能：会话隔离、窗口裁剪、防止Token爆炸
@dataclass
class SessionState:
    session_id: str  # 会话唯一标识，用于区分不同用户（会话隔离）
    history: List[BaseMessage] = field(default_factory=list)  # 对话历史列表

    # 添加消息并保持窗口大小（防止Token爆炸）
    def add(self, msg: BaseMessage, window: int = 6) -> None:
        self.history.append(msg)  # 添加新消息
        self.history = self.history[-window:]  # 只保留最近N轮对话（窗口裁剪）


# 🟣 意图识别层（对应架构图第2层）：根据用户输入判断要执行哪个功能
# 功能：Few-shot提示、多意图解析
def route_intent(user_input: str) -> str:
    s = user_input.lower()
    if "物流" in s or "快递" in s:
        return "track_shipping"  # 物流查询意图
    if "地址" in s:
        return "change_address"  # 修改地址意图
    return "general_qa"  # 默认问答意图


# 🟡 知识库层（对应架构图底层）：每个意图对应的实际业务逻辑
# 功能：FAQ文档、业务规则、工具调用
TOOLS: Dict[str, Callable[[str], str]] = {
    "track_shipping": lambda x: "物流：已出库，运输中（示例）",  # 物流查询工具
    "change_address": lambda x: "改地址：请提供订单号与新地址（示例）",  # 改地址工具
}


# 🟠 路由分发层（对应架构图第3层）：调用对应的工具函数并返回结果
# 功能：条件分支、链路编排
def tool_step(inp: dict) -> dict:
    intent = inp["intent"]  # 获取识别出的意图
    tool = TOOLS.get(intent)  # 获取对应的工具函数
    return {**inp, "tool_result": tool(inp["input"]) if tool else ""}  # 执行工具并返回结果


# 🔵 提示词模板层（连接所有层）：定义给模型的指令格式
# 功能：将路由结果、记忆、工具结果组装成统一的prompt
PROMPT = ChatPromptTemplate.from_messages([
    ("system", "你是电商客服助手。先基于工具结果回答；没有工具结果就正常回答。"),  # 系统角色定义
    MessagesPlaceholder("history"),  # 对话历史占位符（来自记忆管理层）
    ("human", "用户输入：{input}\n意图：{intent}\n工具结果：{tool_result}"),  # 用户输入模板
])

# 🔴 模型执行层（对应架构图第5层）：大语言模型配置
# 功能：通义千问、流式输出
LLM = ChatTongyi(model="qwen-turbo", temperature=0.3)  # 温度越低回答越稳定，客服场景建议 0.2~0.4


# 🔗 完整链路组装：将所有层串联成可插拔流水线
# 对应架构图的数据流向：用户界面层 → 意图识别层 → 路由分发层 → 记忆管理层 → 模型执行层
def build_chain(state: SessionState):
    # 第一步：意图路由器，将用户输入转换为包含意图和历史的数据结构
    # 这里融合了：用户输入（界面层）+ 意图识别（第2层）+ 记忆管理（第4层）
    router = RunnableLambda(lambda x: {"input": x, "intent": route_intent(x), "history": state.history})
    
    # 完整链路：路由 → 工具执行 → 提示词模板 → 大语言模型
    # 数据流：用户输入 → 意图识别 → 路由分发 → 工具调用 → 提示词组装 → 模型推理
    return router | RunnableLambda(tool_step) | PROMPT | LLM


# 🎯 演示执行流程：展示完整的对话流水线
if __name__ == "__main__":
    # 检查API密钥是否配置
    if not os.getenv("DASHSCOPE_API_KEY"):
        raise ValueError("请先设置环境变量 DASHSCOPE_API_KEY")

    # 创建会话状态和对话链路
    s = SessionState(session_id="demo")  # 初始化会话（记忆管理层）
    chain = build_chain(s)  # 构建对话链路（组装所有层）

    # 第一轮对话：演示"用户界面层 → 意图识别层 → 路由分发层 → 模型执行层"
    u1 = "快递到哪了？"  # 用户输入（界面层）
    a1 = chain.invoke(u1)  # 执行完整链路获取回答
    s.add(HumanMessage(content=u1))  # 添加用户消息到历史（记忆管理层）
    s.add(AIMessage(content=a1.content))  # 添加AI回答到历史（记忆管理层）

    # 第二轮对话：演示跨轮上下文（记忆管理层的作用）
    u2 = "那我想改地址"  # 用户追问（界面层）
    a2 = build_chain(s).invoke(u2)  # 重新构建链路：把更新后的 Memory 注入进去
    print(a2.content)

# 完整会话管理实现：窗口裁剪 + 关键事实 + 对话摘要
# 这段代码对应“会话管理的 5 步流水线”：
# - Step 1：调用方传入 session_id（隔离）
# - Step 2：读上下文（Facts/Summary/Window）
# - Step 3：拼 Prompt（按优先级注入，控制预算）
# - Step 4：调用模型得到回复
# - Step 5：写回存储（追加消息、裁剪窗口、更新摘要/事实；并发时需要原子性）

from dataclasses import dataclass, field
from typing import Dict, List, Optional

from langchain_core.messages import AIMessage, BaseMessage, HumanMessage
from langchain_community.chat_models import ChatTongyi


@dataclass
class SessionFacts:
    """Facts：结构化关键事实（优先注入 Prompt，保证关键字段不丢失）"""

    order_id: Optional[str] = None      # 订单号：物流/改地址等场景必须引用
    address: Optional[str] = None       # 地址：改地址场景必须引用
    user_name: Optional[str] = None     # 用户名：个性化回复
    last_intent: Optional[str] = None   # 上次意图：帮助多轮对话保持连贯


@dataclass
class SessionMemory:
    """会话记忆三件套：Facts + Window + Summary

    - Facts：结构化字段（硬信息）
    - Window：最近 N 轮原始对话（追问可用）
    - Summary：更早对话的压缩文本（控预算）
    """

    facts: SessionFacts = field(default_factory=SessionFacts)
    window_messages: List[BaseMessage] = field(default_factory=list)
    summary: str = ""

    def __post_init__(self):
        # Window 的轮数：每轮通常包含（用户 + 助手）两条消息
        self.window_size = 5

    def add_message(self, message: BaseMessage) -> None:
        """Step 5：写回存储（单机内存版）。

        追加消息后进行窗口裁剪：
        - 超出的旧消息进摘要（Summary）
        - 只保留最近 N 轮在 Window
        """

        self.window_messages.append(message)

        # Window 裁剪：只保留最近 N 轮（2N 条消息）
        max_len = self.window_size * 2
        if len(self.window_messages) > max_len:
            overflow = self.window_messages[:-max_len]
            self._update_summary(overflow)
            self.window_messages = self.window_messages[-max_len:]

    def _update_summary(self, old_messages: List[BaseMessage]) -> None:
        """把更早的对话压缩到 Summary，并尽量提取关键事实写入 Facts。"""

        key_info: List[str] = []

        for msg in old_messages:
            content = (msg.content or "").lower()

            # 订单号提取：只在 facts.order_id 为空时写入（避免覆盖已有值）
            if "订单" in content and any(c.isdigit() for c in content):
                import re

                m = re.search(r"\d+", content)
                if m and not self.facts.order_id:
                    self.facts.order_id = m.group()
                    key_info.append(f"订单{m.group()}")

            # 地址提取：简化示例，优先抓城市/门牌等片段
            if "地址" in content or "改地址" in content:
                import re

                patterns = [
                    r"(北京|上海|广州|深圳|杭州|南京|武汉|成都|重庆|西安|天津)[^\s，。！？]*",
                    r"[0-9]+号[^\s，。！？]*",
                ]
                for p in patterns:
                    m = re.search(p, content)
                    if m and not self.facts.address:
                        self.facts.address = m.group()
                        key_info.append(f"地址：{m.group()}")
                        break

        if key_info:
            self.summary += ("；".join(key_info) + "。")

    def get_context_prompt(self) -> str:
        """Step 3：拼 Prompt 时使用。

        只拼 Facts + Summary；Window 原文在拼 prompt 时单独注入。
        """

        parts: List[str] = []
        if self.facts.order_id:
            parts.append(f"订单号：{self.facts.order_id}")
        if self.facts.address:
            parts.append(f"地址：{self.facts.address}")
        if self.facts.user_name:
            parts.append(f"用户：{self.facts.user_name}")
        if self.summary:
            parts.append(f"历史摘要：{self.summary}")

        return "上下文：" + " | ".join(parts) if parts else ""


class MemoryManager:
    """会话管理器（单机内存演示版）。

    Step 1 的隔离点：不同 session_id 对应不同的 SessionMemory。
    生产环境中：建议把 _sessions 换成 Redis/DB。
    """

    def __init__(self):
        self._sessions: Dict[str, SessionMemory] = {}

    def get_session(self, session_id: str) -> SessionMemory:
        if session_id not in self._sessions:
            self._sessions[session_id] = SessionMemory()
        return self._sessions[session_id]

    def update_facts_from_intent(self, session_id: str, intent_result: dict) -> None:
        """把意图识别出来的槽位写入 Facts（结构化字段）。"""

        session = self.get_session(session_id)
        slots = intent_result.get("slots", {})

        if slots.get("order_id"):
            session.facts.order_id = slots["order_id"]
        if slots.get("new_address"):
            session.facts.address = slots["new_address"]
        if slots.get("user_name"):
            session.facts.user_name = slots["user_name"]

        session.facts.last_intent = str(intent_result.get("intent", []))


memory = MemoryManager()


def simple_intent_recognizer(user_input: str) -> dict:
    """意图识别的简化示例（生产中建议用 LLM/规则引擎/分类器）。"""

    intent: List[str] = []
    slots: Dict[str, str] = {}

    if "订单" in user_input:
        import re

        m = re.search(r"\d+", user_input)
        if m:
            intent.append("order_query")
            slots["order_id"] = m.group()

    if "地址" in user_input:
        intent.append("address_change")

    return {"intent": intent, "slots": slots, "confidence": 0.8}


def chat_with_memory(user_input: str, session_id: str) -> str:
    """端到端对话（把 5 步串起来）。"""

    # Step 1：session_id 由调用方传入（隔离）

    # Step 2/5：先做意图识别并把槽位写入 Facts
    intent_result = simple_intent_recognizer(user_input)
    memory.update_facts_from_intent(session_id, intent_result)

    # Step 2：读取会话上下文
    session = memory.get_session(session_id)

    # Step 5：写回用户消息
    session.add_message(HumanMessage(content=user_input))

    # Step 3：拼 Prompt（Facts/Summary + Window）
    context = session.get_context_prompt()
    history_prompt = ""
    for msg in session.window_messages[:-1]:
        role = "用户" if isinstance(msg, HumanMessage) else "助手"
        history_prompt += f"{role}：{msg.content}\n"

    full_prompt = f"""你是电商客服助手。

{context}

{history_prompt}

用户：{user_input}
助手："""

    # Step 4：调用模型得到回复
    model = ChatTongyi(model="qwen-turbo", temperature=0.3)
    response = model.invoke(full_prompt)

    # Step 5：写回助手消息（并触发 Window 裁剪 / Summary 更新）
    session.add_message(AIMessage(content=response.content))

    return response.content


if __name__ == "__main__":
    # 简单演示：同一个 session_id 连续对话；换 session_id 就不会串话
    session_id = "user_001"

    print("用户：我的订单 12345 想改地址")
    print("助手：", chat_with_memory("我的订单 12345 想改地址", session_id))

    print("\n用户：我刚才那个订单还能改吗？")
    print("助手：", chat_with_memory("我刚才那个订单还能改吗？", session_id))

# 主流程（可直接套进你的项目）

# Client：每次发送都生成 message_id；失败自动重试（指数退避）
message_id = uuid4()
for attempt in [0, 1, 2, 3]:
    try:
        resp = POST("/chat", json={"session_id": sid, "message_id": message_id, "input": text})
        return resp
    except NetworkError:
        sleep(0.3 * (2 ** attempt))

# Server：同一 message_id 只允许处理一次；重复请求直接返回第一次结果
if cache.exists("resp:{sid}:{message_id}"):
    return cache.get("resp:{sid}:{message_id}")

if not cache.setnx("idem:{sid}:{message_id}", "1", ttl=60):
    return cache.wait_for("resp:{sid}:{message_id}")

result = run_business_logic(text)   # 意图识别/工具调用/写回 Step 5
cache.set("resp:{sid}:{message_id}", result, ttl=3600)
return result

import json
import time

import redis


ADD_AND_TRIM_LUA = """
-- KEYS[1] = session:{tenant}:{user}
-- ARGV[1] = message_json
-- ARGV[2] = window_size
-- ARGV[3] = ttl_seconds
-- ARGV[4] = now_ts
-- 伪代码：GET -> APPEND -> TRIM -> UPDATE(summary/facts) -> EXPIRE
return 1
"""


class RedisSessionStore:
    def __init__(self, r: redis.Redis):
        self.r = r
        self.add_and_trim = self.r.register_script(ADD_AND_TRIM_LUA)

    def append_message_atomic(
        self,
        session_key: str,
        msg: dict,
        window_size: int,
        ttl_seconds: int,
    ) -> None:
        self.add_and_trim(
            keys=[session_key],
            args=[
                json.dumps(msg, ensure_ascii=False),
                str(window_size),
                str(ttl_seconds),
                str(int(time.time())),
            ],
        )

# 先结构化，再回答：Intent(JSON) → Router → Tool/Answer
#
# 这一段示例要解决的核心问题：
# 1）模型先输出“可执行 JSON 合同”，而不是直接自由回答（避免编造/跑题）
# 2）你再基于 JSON 做：路由、补槽、兜底

import json
import os
import re
from typing import Any, Dict, List, Literal, Optional

from langchain_community.chat_models import ChatTongyi
from langchain_core.prompts import ChatPromptTemplate
from pydantic import BaseModel, Field, ValidationError


# ===============================
# 1) JSON 合同（Intent Contract）
# ===============================

IntentName = Literal[
    "track_shipping",  # 查物流
    "change_address",  # 改地址
    "refund",          # 退货退款
    "general",         # 其他（兜底）
]


class IntentSlots(BaseModel):
    """槽位：用于补全任务执行所需的关键参数。

    约定：
    - 缺失时填 '?'（字符串），便于下游做“补槽追问”。
    - 即使不需要也可以为空，但字段存在更利于统一处理。
    """

    order_id: Optional[str] = None
    new_address: Optional[str] = None


class IntentResult(BaseModel):
    """模型输出的意图识别结果（严格 JSON）。"""

    intent: List[IntentName] = Field(default_factory=list)
    slots: IntentSlots = Field(default_factory=IntentSlots)
    confidence: float = Field(ge=0.0, le=1.0)


# ===============================
# 2) Prompt：强约束 + few-shot
# ===============================

INTENT_PROMPT = ChatPromptTemplate.from_messages(
    [
        (
            "system",
            "你是电商客服【意图识别器】。你必须只输出严格 JSON（不要 Markdown，不要解释）。\n"
            "要求：\n"
            "- 支持多意图（intent 是数组）。\n"
            "- intent 只能取：track_shipping/change_address/refund/general。\n"
            "- slots 至少包含：order_id/new_address。缺失填 '?'。\n"
            "- confidence 为 0~1 的小数。\n",
        ),
        (
            "human",
            "用户输入：我想改地址，顺便问下快递到哪了？\n"
            "只输出 JSON：\n"
            '{{"intent":["track_shipping","change_address"],"slots":{{"order_id":"?","new_address":"?"}},"confidence":0.86}}',
        ),
        (
            "human",
            "用户输入：订单12345到哪了\n"
            "只输出 JSON：\n"
            '{{"intent":["track_shipping"],"slots":{{"order_id":"12345","new_address":"?"}},"confidence":0.92}}',
        ),
        (
            "human",
            "用户输入：我想退货退款\n"
            "只输出 JSON：\n"
            '{{"intent":["refund"],"slots":{{"order_id":"?","new_address":"?"}},"confidence":0.80}}',
        ),
        (
            "human",
            "用户输入：{input}\n"
            "只输出 JSON：",
        ),
    ]
)


# ===============================
# 3) JSON 提取与容错（非常关键）
# ===============================

def extract_json(text: str) -> str:
    """从模型输出中提取 JSON。

    现实情况：模型可能会返回 ```json ... ``` 或夹杂多余解释。
    我们尽量从中抽取 { ... } 这一段。
    """

    t = (text or "").strip()
    t = re.sub(r"^```(?:json)?\s*", "", t, flags=re.IGNORECASE)
    t = re.sub(r"\s*```$", "", t)
    m = re.search(r"\{.*\}", t, flags=re.S)
    return m.group(0) if m else t


def safe_load_intent(text: str) -> IntentResult:
    """把模型输出解析为 IntentResult。

    出错时抛异常，由上层统一兜底（比如转人工、或提示用户换问法）。
    """

    raw = json.loads(extract_json(text))
    return IntentResult.model_validate(raw)


# ===============================
# 4) 识别函数（把 LLM 封装成可复用组件）
# ===============================

def recognize_intent(user_input: str) -> IntentResult:
    llm = ChatTongyi(model="qwen-turbo", temperature=0)
    resp = (INTENT_PROMPT | llm).invoke({"input": user_input})
    return safe_load_intent(resp.content)


# ===============================
# 5) Router + 补槽 + 兜底
# ===============================

CONFIDENCE_THRESHOLD = 0.65


def need_slot(value: Optional[str]) -> bool:
    return (value is None) or (str(value).strip() in {"", "?"})


def route_and_act(ir: IntentResult) -> str:
    """根据识别结果执行策略。

    这里演示三件事：
    - 低置信度兜底
    - 缺槽位追问（补槽）
    - 多意图路由（先做一个，再做另一个）
    """

    if ir.confidence < CONFIDENCE_THRESHOLD or not ir.intent:
        return "抱歉，我没太理解。你是想：1）查物流进度 2）改收货地址 3）退货退款？请告诉我具体需求。"

    replies: List[str] = []

    for intent in ir.intent:
        if intent == "track_shipping":
            if need_slot(ir.slots.order_id):
                replies.append("我可以帮你查物流进度。请先提供订单号。")
            else:
                replies.append(f"好的，我来帮你查询订单 {ir.slots.order_id} 的物流进度。")

        elif intent == "change_address":
            if need_slot(ir.slots.order_id):
                replies.append("我可以帮你改地址。请先提供订单号。")
            elif need_slot(ir.slots.new_address):
                replies.append("好的，订单号已收到。请提供新的收货地址（省市区+详细地址）。")
            else:
                replies.append(f"好的，我将把订单 {ir.slots.order_id} 的地址修改为：{ir.slots.new_address}。")

        elif intent == "refund":
            if need_slot(ir.slots.order_id):
                replies.append("我可以协助你退货退款。请先提供订单号。")
            else:
                replies.append(f"好的，我来为订单 {ir.slots.order_id} 发起退货退款流程。")

        else:
            replies.append("我可以继续为你解答。你想咨询哪方面？")

    return "\n".join(replies)


if __name__ == "__main__":
    if not os.getenv("DASHSCOPE_API_KEY"):
        raise ValueError("请先设置环境变量 DASHSCOPE_API_KEY")

    user_question = "我想改地址，顺便问下快递到哪了？"

    try:
        ir = recognize_intent(user_question)
        print("意图识别结果：", ir.model_dump())
        print("\n对用户回复：\n", route_and_act(ir))
    except (ValidationError, json.JSONDecodeError) as e:
        raise RuntimeError(f"intent_contract_error: {e}")
    except Exception as e:
        raise RuntimeError(f"intent_runtime_error: {e}")

# 流式输出实现 - 四步结构化输出法（真实调用大模型）
# 保存为 utils.py，在 main.py 中调用

import os
from typing import Generator

from langchain_community.chat_models import ChatTongyi
from langchain_core.output_parsers import StrOutputParser
from langchain_core.prompts import ChatPromptTemplate


PROMPT = ChatPromptTemplate.from_template(
    """你是电商客服助手。请用【四步结构化输出】回答，并保持顺序：
1) 先结论：一句话结论（首屏要短）
2) 再步骤：用 3~6 条可执行步骤
3) 再注意事项：列出 2~4 条风险/限制/坑
4) 给下一步：给出一句追问或下一步动作（如索要订单号）

注意：回答时要自然，不要把“先结论：”、“再步骤：”、“再注意事项：”、“给下一步：”这些标签直接写进回复里。分条的部分请用换行分隔。

用户问题：{input}
"""
)


def structured_stream_response(user_input: str, model: ChatTongyi | None = None) -> Generator[str, None, None]:
    """真实流式输出：直接从大模型 stream 出 token。

    关键点：
    - 不再用 time.sleep 模拟“打字”，而是用模型的流式能力。
    - 首 token 时间取决于模型与网络，一般可做到 < 1s。
    """

    if not os.getenv("DASHSCOPE_API_KEY"):
        raise ValueError("请先设置环境变量 DASHSCOPE_API_KEY")

    llm = model or ChatTongyi(model="qwen-turbo", temperature=0.3)
    chain = PROMPT | llm | StrOutputParser()

    for chunk in chain.stream({"input": user_input}):
        yield chunk


def demo_usage():
    """终端演示：可观察逐步输出。"""
    print("用户：退货怎么走？")
    print("AI：", end="", flush=True)
    for chunk in structured_stream_response("退货怎么走？"):
        print(chunk, end="", flush=True)
    print()


if __name__ == "__main__":
    demo_usage()

import json
import time
import uuid
from dataclasses import dataclass, asdict
from typing import Any, Dict, Optional


@dataclass
class Span:
    name: str
    start_ms: int
    end_ms: int

    @property
    def latency_ms(self) -> int:
        return self.end_ms - self.start_ms


def now_ms() -> int:
    return int(time.time() * 1000)


def new_trace_id() -> str:
    return uuid.uuid4().hex


def mask_text(text: str) -> str:
    """示意：生产中请用更严格的脱敏规则。"""
    if not text:
        return ""
    return text.replace("88号", "**号")


def log_turn(
    *,
    trace_id: str,
    session_id: str,
    turn_id: int,
    entry_channel: str,
    model: str,
    prompt_version: str,
    prompt_hash: str,
    user_input: str,
    intent: Optional[Any] = None,
    slots: Optional[Dict[str, Any]] = None,
    output: str = "",
    tokens_in: Optional[int] = None,
    tokens_out: Optional[int] = None,
    spans: Optional[Dict[str, Span]] = None,
    error_type: Optional[str] = None,
) -> None:
    total_ms = None
    if spans and "total" in spans:
        total_ms = spans["total"].latency_ms

    event = {
        "trace_id": trace_id,
        "session_id": session_id,
        "turn_id": turn_id,
        "entry_channel": entry_channel,
        "model": model,
        "prompt_version": prompt_version,
        "prompt_hash": prompt_hash,
        "input": mask_text(user_input),
        "intent": intent,
        "slots": slots or {},
        "output": mask_text(output),
        "tokens_in": tokens_in,
        "tokens_out": tokens_out,
        "latency_ms_total": total_ms,
        "latency_ms_intent": spans.get("intent").latency_ms if spans and spans.get("intent") else None,
        "latency_ms_tool": spans.get("tool").latency_ms if spans and spans.get("tool") else None,
        "latency_ms_llm": spans.get("llm").latency_ms if spans and spans.get("llm") else None,
        "error_type": error_type,
    }
    print(json.dumps(event, ensure_ascii=False))


def demo_log():
    trace_id = new_trace_id()
    spans = {}
    spans["total"] = Span(name="total", start_ms=now_ms(), end_ms=now_ms())
    log_turn(
        trace_id=trace_id,
        session_id="abc",
        turn_id=3,
        entry_channel="http_api",
        model="qwen-turbo",
        prompt_version="v12",
        prompt_hash="c0ffee...",
        user_input="我想改地址，新地址是北京市朝阳区建国路88号",
        intent=["change_address"],
        slots={"order_id": "?", "new_address": "北京市朝阳区建国路**号"},
        output="好的，请先提供订单号。",
        tokens_in=520,
        tokens_out=80,
        spans=spans,
        error_type=None,
    )


if __name__ == "__main__":
    demo_log()

# streamlit_app.py —— UI 入口只做“适配”，核心逻辑全部在 utils.py

import streamlit as st

from utils import get_chat_response
from utils_protocol import ChatRequest, build_session_id, new_trace_id


st.title("电商客服")

# 入口层：收集输入（不同入口收集方式不同，但最终都要转成统一协议 ChatRequest）
user_input = st.chat_input("请输入你的问题")

if user_input:
    req = ChatRequest(
        trace_id=new_trace_id(),
        session_id=build_session_id(
            entry_channel="streamlit",
            tenant_id="demo_tenant",
            user_id="demo_user",
        ),
        entry_channel="streamlit",
        user_input=user_input,
        metadata={"page": "chapter3"},
    )

    # 核心层：只依赖统一协议，不关心入口形态
    reply = get_chat_response(req)
    st.chat_message("assistant").write(reply)

# api_server.py —— HTTP 入口只做：鉴权/限流/适配协议，然后调用 utils.get_chat_response

from fastapi import FastAPI, Header
from pydantic import BaseModel

from utils import get_chat_response
from utils_protocol import ChatRequest, build_session_id, new_trace_id


app = FastAPI()


class HttpChatBody(BaseModel):
    user_input: str
    tenant_id: str
    user_id: str


@app.post("/chat")
def chat(body: HttpChatBody, x_api_key: str = Header(default="")):
    # 入口层常见职责：鉴权（示意，不要在代码里硬编码 key）
    if not x_api_key:
        return {"error": "missing_api_key"}

    req = ChatRequest(
        trace_id=new_trace_id(),
        session_id=build_session_id(
            entry_channel="http_api",
            tenant_id=body.tenant_id,
            user_id=body.user_id,
        ),
        entry_channel="http_api",
        user_input=body.user_input,
        metadata={"ip": "...", "ua": "..."},
    )

    reply = get_chat_response(req)
    return {"trace_id": req.trace_id, "session_id": req.session_id, "reply": reply}


# 启动：uvicorn api_server:app --port 8000

# webhook_adapter.py —— 回调入口只做“事件解析 → 协议适配 → 调核心 → 返回结果”

from typing import Dict

from utils import get_chat_response
from utils_protocol import ChatRequest, build_session_id, new_trace_id


def handle_wecom_event(event: Dict) -> Dict:
    """企业微信回调适配示例（伪代码）。

    event 里通常会有：
    - tenant_id / corp_id
    - user_id / open_id
    - msg_type / content
    """

    tenant_id = event.get("corp_id", "")
    user_id = event.get("open_id", "")
    user_input = event.get("content", "")

    req = ChatRequest(
        trace_id=new_trace_id(),
        session_id=build_session_id(
            entry_channel="wecom",
            tenant_id=tenant_id,
            user_id=user_id,
        ),
        entry_channel="wecom",
        user_input=user_input,
        metadata={"msg_type": event.get("msg_type")},
    )

    reply = get_chat_response(req)
    # 入口层负责把 reply 转回渠道需要的格式
    return {"msgtype": "text", "text": {"content": reply}}