第七章 提示词链与任务分解¶

def process_query(user_input):
    # 步骤 1：意图分类
    intent = classify_intent(user_input)

    # 步骤 2：分支处理
    if intent == "product_inquiry":
        response = handle_product_inquiry(user_input)
    elif intent == "order_status":
        response = handle_order_status(user_input)
    elif intent == "complaint":
        response = handle_complaint(user_input)
    else:
        response = handle_general(user_input)

    return response

result1 = {"summary": "（示例）步骤 1 的输出结果"}

state = {
    "original_input": "...",
    "step1_result": None,
    "step2_result": None,
    "errors": []
}

# 步骤 1 执行后
state["step1_result"] = result1

# 步骤 2 可以访问
step2_input = f"""
原始输入：{state["original_input"]}
步骤 1 结果：{state["step1_result"]}
当前任务：...
"""

print(step2_input.strip())

class InMemoryStorage:
    def __init__(self):
        self._data = {}

    def save(self, session_id, key, value):
        self._data[(session_id, key)] = value

    def load(self, session_id, key):
        return self._data[(session_id, key)]

storage = InMemoryStorage()
session_id = "demo-session"
result1 = {"summary": "（示例）步骤 1 的输出结果"}

# 步骤 1 完成后

storage.save(session_id, "step1_result", result1)

# 步骤 2 需要时

step1_result = storage.load(session_id, "step1_result")

print(step1_result)

class Step1:
    input_schema = {
        "text": str,
        "config": dict
    }
    output_schema = {
        "entities": list,
        "sentiment": str,
        "confidence": float
    }

def build_step2_prompt(step1_output, step2_config):
    return f"""

## 上一步结果

{json.dumps(step1_output, ensure_ascii=False)}

## 当前任务

{step2_config['task_description']}

## 输出要求

{step2_config['output_format']}
"""

def execute_chain(steps, initial_input):
    current_input = initial_input
    results = []

    for i, step in enumerate(steps):
        try:
            output = step.execute(current_input)
            results.append({
                "step": i + 1,
                "status": "success",
                "output": output
            })
            current_input = output
        except Exception as e:
            results.append({
                "step": i + 1,
                "status": "error",
                "error": str(e)
            })
            # 决定是否继续
            if step.critical:
                break

    return results

class CompetitorAnalysisWorkflow:

    def __init__(self, llm_client):
        self.llm = llm_client

    def run(self, product_info):
        # 步骤 1：信息整理
        step1_result = self.execute_step(
            "信息整理",
            STEP1_PROMPT.format(raw_product_info=product_info)
        )

        # 步骤 2 和 3：并行执行
        step2_result = self.execute_step(
            "功能对比",
            STEP2_PROMPT.format(step1_output=step1_result)
        )
        step3_result = self.execute_step(
            "定价分析",
            STEP3_PROMPT.format(step1_output=step1_result)
        )

        # 步骤 4：SWOT 分析
        step4_result = self.execute_step(
            "SWOT 分析",
            STEP4_PROMPT.format(
                step2_output=step2_result,
                step3_output=step3_result
            )
        )

        # 步骤 5：报告生成
        step5_result = self.execute_step(
            "报告生成",
            STEP5_PROMPT.format(
                step1_output=step1_result,
                step2_output=step2_result,
                step3_output=step3_result,
                step4_output=step4_result
            )
        )

        # 步骤 6：质量检查（带重试）
        final_report = self.quality_check_with_retry(step5_result)

        return final_report

    def execute_step(self, step_name, prompt):
        print(f"执行步骤: {step_name}")
        response = self.llm.generate(prompt)
        return response

    def quality_check_with_retry(self, report, max_retries=2):
        for attempt in range(max_retries + 1):
            # 步骤 6 的提示词要求返回 JSON，这里必须解析后才能取字段
            check_result = json.loads(self.execute_step(
                "质量检查",
                STEP6_PROMPT.format(step5_output=report)
            ))

            if check_result.get("approved", False):
                return report

            # 如果未通过，进行修正
            report = self.revise_report(
                report,
                check_result.get("suggestions", [])
            )

        return report

    def revise_report(self, report, suggestions):
        """按质检意见修订报告"""
        suggestion_text = "\n".join(suggestions)
        return self.execute_step(
            "报告修订",
            f"请根据以下修改建议修订这份竞品分析报告。\n\n"
            f"## 修改建议\n{suggestion_text}\n\n"
            f"## 原报告\n{report}"
        )

class SlidingWindowDialogueManager:
    """滑动窗口对话管理器"""

    def __init__(self, window_size: int = 10, system_prompt: str = ""):
        self.window_size = window_size
        self.system_prompt = system_prompt
        self.full_history = []

    def add_message(self, role: str, content: str):
        """添加消息到对话历史"""
        self.full_history.append({
            "role": role,
            "content": content,
            "timestamp": self._get_timestamp()
        })

    def get_context_for_api(self) -> List[Dict]:
        """获取用于 API 调用的上下文

        包含系统提示词和最近 N 轮对话
        """

        # 总是包含系统提示词
        context = [{"role": "system", "content": self.system_prompt}]

        # 添加最近 N 轮对话
        recent_messages = self.full_history[-self.window_size * 2:]
        context.extend(recent_messages)

        return context

    def get_stats(self) -> Dict:
        """获取对话统计信息"""
        return {
            "total_messages": len(self.full_history),
            "window_messages": min(self.window_size * 2, len(self.full_history)),
            "estimated_tokens": self._estimate_tokens(self.get_context_for_api()),
            "api_history_size": len(self.full_history),
            "context_loss": self._get_loss_percentage()
        }

    def _estimate_tokens(self, messages: List[Dict]) -> int:
        """估算 Token 数量（粗略）"""
        # 简化估算：1 个 token ≈ 4 个字符
        total_chars = sum(len(m.get("content", "")) for m in messages)
        return total_chars // 4

    def _get_loss_percentage(self) -> float:
        """计算因窗口截断而丢失的信息百分比"""
        total = len(self.full_history)
        kept = min(self.window_size * 2, total)
        return (total - kept) / total if total > 0 else 0

    def _get_timestamp(self) -> str:
        from datetime import datetime
        return datetime.now().isoformat()


# 使用示例

manager = SlidingWindowDialogueManager(window_size=5)
manager.system_prompt = "你是一个旅游顾问。"

# 模拟一个长对话

for i in range(20):
    manager.add_message("user", f"问题 {i}")
    manager.add_message("assistant", f"回答 {i}")

# 获取用于 API 的上下文

api_context = manager.get_context_for_api()
print(f"API 调用将包含 {len(api_context)} 条消息")
print(f"预估 Token: {manager.get_stats()['estimated_tokens']}")
print(f"丢失信息: {manager.get_stats()['context_loss']:.1%}")

class DynamicSummarizationManager:
    """动态摘要对话管理器"""

    def __init__(
        self,
        llm_client,
        max_context_tokens: int = 3000,
        system_prompt: str = ""
    ):
        self.llm = llm_client
        self.max_context_tokens = max_context_tokens
        self.system_prompt = system_prompt
        self.messages = []
        self.summaries = []  # 存储历史摘要

    def add_message(self, role: str, content: str):
        """添加消息"""
        self.messages.append({
            "role": role,
            "content": content,
            "timestamp": self._get_timestamp()
        })

        # 检查是否需要摘要
        if self._should_summarize():
            self._trigger_summarization()

    def _should_summarize(self) -> bool:
        """检查是否需要进行摘要"""
        current_tokens = self._estimate_tokens(self.messages)
        return current_tokens > self.max_context_tokens

    def _trigger_summarization(self):
        """触发摘要过程"""

        # 获取需要摘要的消息（保留最后 10 条）
        to_summarize = self.messages[:-10]
        remaining = self.messages[-10:]

        if len(to_summarize) < 4:
            return  # 太短，不需要摘要

        # 构建摘要提示词
        summary_prompt = self._build_summary_prompt(to_summarize)

        # 调用 LLM 进行摘要
        summary = self.llm.generate(summary_prompt)

        # 存储摘要
        self.summaries.append({
            "content": summary,
            "messages_count": len(to_summarize),
            "timestamp": self._get_timestamp()
        })

        # 替换消息
        self.messages = [
            {
                "role": "system",
                "content": f"[早期对话摘要] {summary}"
            }
        ] + remaining

    def _build_summary_prompt(self, messages: List[Dict]) -> str:
        """构建摘要提示词"""

        conversation_text = "\n".join([
            f"{m['role']}: {m['content']}"
            for m in messages
        ])

        return f"""请总结以下对话的核心内容和关键信息，用 1-2 句话表达。
总结应该保留用户的主要需求、已做的决定和重要的上下文信息。

对话内容：
{conversation_text}

摘要："""

    def get_context_for_api(self) -> List[Dict]:
        """获取用于 API 调用的上下文"""
        context = [{"role": "system", "content": self.system_prompt}]
        context.extend(self.messages)
        return context

    def get_stats(self) -> Dict:
        """获取统计信息"""
        return {
            "total_messages": len(self.messages),
            "summaries_count": len(self.summaries),
            "estimated_tokens": self._estimate_tokens(self.get_context_for_api()),
            "compression_ratio": self._calculate_compression_ratio()
        }

    def _estimate_tokens(self, messages: List[Dict]) -> int:
        total_chars = sum(len(m.get("content", "")) for m in messages)
        return total_chars // 4

    def _calculate_compression_ratio(self) -> float:
        """计算压缩比"""
        total_summarized = sum(s["messages_count"] for s in self.summaries)
        total_current = len(self.messages)
        return total_summarized / (total_summarized + total_current) if total_summarized > 0 else 0

    def _get_timestamp(self) -> str:
        from datetime import datetime
        return datetime.now().isoformat()


# 使用示例

manager = DynamicSummarizationManager(
    llm_client=openai_client,
    max_context_tokens=3000
)

# 长对话模拟

for i in range(50):
    manager.add_message("user", f"用户问题 {i}: 详细的问题内容...")
    manager.add_message("assistant", f"助手回答 {i}: 详细的回答内容...")

stats = manager.get_stats()
print(f"摘要数: {stats['summaries_count']}")
print(f"压缩比: {stats['compression_ratio']:.1%}")

from dataclasses import dataclass, asdict
from typing import Optional
from datetime import datetime

@dataclass
class DialogueState:
    """对话状态快照"""
    user_id: str
    conversation_id: str
    intent: Optional[str] = None  # 用户的主要意图
    entities: dict = None  # 关键实体
    sentiment: Optional[str] = None  # 情感状态
    stage: Optional[str] = None  # 对话阶段
    last_updated: str = None

    def __post_init__(self):
        if self.entities is None:
            self.entities = {}
        if self.last_updated is None:
            self.last_updated = datetime.now().isoformat()


class StatefulDialogueManager:
    """有状态对话管理器"""

    def __init__(self, llm_client, db_client, state_schema: dict):
        self.llm = llm_client
        self.db = db_client  # 数据库客户端
        self.state_schema = state_schema  # 定义哪些字段需要追踪
        self.current_state = None

    def process_turn(self, user_input: str, conversation_id: str) -> str:
        """处理一轮对话"""

        # 1. 加载或初始化状态
        self.current_state = self._load_state(conversation_id)

        # 2. 使用 LLM 提取关键信息
        extracted = self._extract_entities(user_input)

        # 3. 更新状态
        self._update_state(extracted)

        # 4. 根据当前状态生成响应
        response = self._generate_response(user_input)

        # 5. 持久化状态
        self._save_state(conversation_id)

        return response

    def _extract_entities(self, user_input: str) -> dict:
        """使用 LLM 提取实体"""

        extraction_prompt = f"""从以下用户输入中提取关键信息。

当前状态：{self.current_state.entities}

用户输入：{user_input}

请提取以下字段（如果不存在则为 null）：
- 用户名
- 订单 ID
- 问题类型
- 优先级
- 情感（positive/negative/neutral）

返回 JSON 格式。"""

        response = self.llm.generate(extraction_prompt)
        # 解析 JSON 响应
        return self._parse_json_response(response)

    def _update_state(self, extracted: dict):
        """更新状态对象"""

        for key, value in extracted.items():
            if key not in self.state_schema:
                continue
            if hasattr(self.current_state, key):
                # sentiment 等已在 DialogueState 中声明的字段，直接写入属性
                setattr(self.current_state, key, value)
            else:
                # 其余字段统一放进 entities，否则 asdict() 不会持久化它们
                self.current_state.entities[key] = value

        self.current_state.last_updated = datetime.now().isoformat()

    def _generate_response(self, user_input: str) -> str:
        """根据当前状态生成响应"""

        # 构建包含状态的提示词
        prompt = f"""你是一个客服助手。

当前用户状态：
- 名字：{self.current_state.entities.get('user_name')}
- 订单 ID：{self.current_state.entities.get('order_id')}
- 问题类型：{self.current_state.entities.get('issue_type')}
- 情感：{self.current_state.sentiment}

用户输入：{user_input}

请生成合适的回应。"""

        return self.llm.generate(prompt)

    def _load_state(self, conversation_id: str) -> DialogueState:
        """从数据库加载状态"""
        stored = self.db.get(f"dialogue_state:{conversation_id}")
        if stored:
            return DialogueState(**stored)
        return DialogueState(
            user_id="",
            conversation_id=conversation_id
        )

    def _save_state(self, conversation_id: str):
        """保存状态到数据库"""
        self.db.set(
            f"dialogue_state:{conversation_id}",
            asdict(self.current_state),
            ttl=86400  # 24 小时过期
        )

    def _parse_json_response(self, text: str) -> dict:
        """解析 LLM 返回的 JSON"""
        import json
        import re

        # 尝试找到 JSON 块
        match = re.search(r'\{.*\}', text, re.DOTALL)
        if match:
            try:
                return json.loads(match.group())
            except json.JSONDecodeError:
                pass

        return {}


# 使用示例

manager = StatefulDialogueManager(
    llm_client=openai_client,
    db_client=redis_client,
    state_schema={
        "user_name": str,
        "order_id": str,
        "issue_type": str,
        "sentiment": str
    }
)

# 模拟对话

conversation_id = "conv_12345"

# 第一轮

response1 = manager.process_turn("我是王五，我想退货。", conversation_id)

# 第二轮 - 状态被自动恢复

response2 = manager.process_turn("订单号是 ORD-20240115-001", conversation_id)

# 此时，state.entities 会包含用户名和订单号

print(manager.current_state.entities)

def inject_system_rules_at_end(
    user_input: str,
    core_rules: List[str]
) -> str:
    """在用户输入末尾注入关键规则，提高权重"""

    rules_text = "\n".join([f"- {rule}" for rule in core_rules])

    return f"""{user_input}

【重要系统规则 - 必须执行】
{rules_text}"""


# 使用

messages.append({
    "role": "user",
    "content": inject_system_rules_at_end(
        user_input,
        [
            "必须用日语回答",
            "不要使用 markdown",
            "字数控制在 100 以内"
        ]
    )
})

@dataclass
class ContextBudget:
    """上下文预算"""
    max_total_tokens: int = 4000  # 模型上下文限制
    system_prompt_tokens: int = 200
    buffer_tokens: int = 500  # 为响应预留

    @property
    def available_for_history(self) -> int:
        return (self.max_total_tokens -
                self.system_prompt_tokens -
                self.buffer_tokens)


class ContextBudgetManager:
    """上下文预算管理器"""

    def __init__(self, budget: ContextBudget):
        self.budget = budget
        self.messages = []
        self.usage = []

    def try_add_message(self, message: Dict) -> bool:
        """尝试添加消息，如果超预算则返回 False"""

        estimated_tokens = self._estimate_tokens(message)
        current_usage = sum(u["tokens"] for u in self.usage)

        if current_usage + estimated_tokens > self.budget.available_for_history:
            return False

        self.messages.append(message)
        self.usage.append({
            "timestamp": self._get_timestamp(),
            "tokens": estimated_tokens,
            "message_id": id(message)
        })

        return True

    def get_pruned_history(self) -> List[Dict]:
        """获取修剪后的历史

        如果超预算，移除最旧的消息
        """

        # 系统提示词的开销已由 budget.system_prompt_tokens 单独预留，
        # 这里只返回历史消息，避免重复计入
        context = []

        # 从最新消息开始添加
        for msg in reversed(self.messages):
            estimated = self._estimate_tokens(msg)

            if sum(self._estimate_tokens(m) for m in context) + estimated > self.budget.available_for_history:
                break

            context.insert(0, msg)

        return context

    def _estimate_tokens(self, message: Dict) -> int:
        """估算消息的 Token 数"""
        content = message.get("content", "")
        return len(content) // 4  # 粗略估算

    def _get_timestamp(self) -> str:
        from datetime import datetime
        return datetime.now().isoformat()


# 使用示例

budget = ContextBudget(
    max_total_tokens=4096,
    system_prompt_tokens=200,
    buffer_tokens=500
)

manager = ContextBudgetManager(budget)

# 添加消息，超预算时拒绝

for msg in dialogue_messages:
    if not manager.try_add_message(msg):
        print(f"超过上下文预算，无法添加更多消息")
        break

# 获取用于 API 的历史

pruned = manager.get_pruned_history()