💬 项目3：多轮对话任务 Agent

难度: ⭐⭐⭐⭐⭐ Claude Code ⏱ 预计 5-6 小时

1. 需求分析

什么是"多轮对话任务 Agent"？

前面的 Agent 大多是"一问一答"模式：用户输入 → Agent 执行 → 返回结果。但真实场景中，很多任务需要多轮交互：

• 用户说"帮我订机票"，Agent 需要追问：从哪里到哪里？什么日期？偏好什么时段？
• 用户说"帮我写个方案"，Agent 需要：先确认方案类型 → 了解需求细节 → 出大纲给用户确认 → 逐章撰写
• 执行过程中发现信息不足 → Agent 主动追问，而非胡乱执行
• 用户中途改变主意 → Agent 能灵活调整，而非死板地按原计划执行

核心挑战

• 🎯 意图识别：用户说的话到底想干什么？新任务还是对之前的修正？
• 📊 状态管理：当前任务进行到哪一步了？已收集了哪些信息？还缺什么？
• 🔄 动态规划：根据用户反馈动态调整计划，而非一次性规划然后蒙头执行
• ⚠️ 异常适应：执行失败或用户中断时，能优雅地恢复或切换

2. 架构设计

3. 完整代码

#!/usr/bin/env python3
"""
MultiTurnAgent — 多轮对话任务 Agent
======================================
核心能力：意图识别 | 状态管理 | 动态规划 | 异常适应

运行：python3 multi_turn_agent.py
"""

import json
import os
from datetime import datetime
from pathlib import Path
from enum import Enum

from dotenv import load_dotenv
from anthropic import Anthropic

load_dotenv()


# ================================================================
# 第1部分：状态定义
# ================================================================

class TaskStatus(Enum):
    IDLE = "idle"               # 空闲，等待新任务
    COLLECTING = "collecting"   # 收集信息中（追问阶段）
    PLANNING = "planning"       # 规划中
    EXECUTING = "executing"     # 执行中
    WAITING_FEEDBACK = "waiting_feedback"  # 等待用户确认
    COMPLETED = "completed"     # 完成
    CANCELLED = "cancelled"     # 已取消


class DialogState:
    """对话状态管理器 — 跟踪当前任务的完整状态"""

    def __init__(self):
        self.reset()

    def reset(self):
        self.status = TaskStatus.IDLE
        self.task_type: str = ""           # 任务类型
        self.task_description: str = ""    # 任务描述
        self.collected_info: dict = {}     # 已收集的信息
        self.missing_info: list[str] = []  # 还缺少的信息
        self.plan: list[dict] = []         # 执行计划
        self.current_step: int = 0         # 当前步骤
        self.execution_results: list = []  # 执行结果
        self.conversation_history: list[dict] = []  # 本任务对话历史
        self.user_intent: str = ""         # 最近一次用户意图
        self.retry_count: int = 0          # 重试次数

    def add_turn(self, user: str, agent: str):
        self.conversation_history.append({
            "user": user, "agent": agent, "time": datetime.now().isoformat()
        })
        # 只保留最近20轮
        if len(self.conversation_history) > 20:
            self.conversation_history = self.conversation_history[-20:]

    def summary(self) -> str:
        return json.dumps({
            "status": self.status.value,
            "task_type": self.task_type,
            "collected": self.collected_info,
            "missing": self.missing_info,
            "current_step": self.current_step,
        }, ensure_ascii=False)


# ================================================================
# 第2部分：意图识别器
# ================================================================

class IntentRecognizer:
    """识别用户在对话中的意图"""

    INTENT_TYPES = [
        "new_task",        # 发起新任务
        "provide_info",    # 提供信息（回答Agent的追问）
        "confirm",         # 确认/同意
        "reject",          # 拒绝/不同意
        "modify",          # 修改之前的指令
        "cancel",          # 取消当前任务
        "question",        # 一般性问题
        "unclear",         # 意图不明确
    ]

    def __init__(self, client: Anthropic, model: str = "claude-sonnet-4-6"):
        self.client = client
        self.model = model

    def recognize(self, user_input: str, current_state: DialogState) -> dict:
        """识别用户意图"""
        state_context = ""
        if current_state.status != TaskStatus.IDLE:
            state_context = f"""
当前任务状态: {current_state.status.value}
当前任务: {current_state.task_description}
已收集信息: {json.dumps(current_state.collected_info, ensure_ascii=False)}
还缺信息: {current_state.missing_info}
"""

        prompt = f"""分析用户在对话中的意图。{state_context}

用户消息: "{user_input}"

请输出 JSON 格式：
{{
  "intent": "意图类型",
  "confidence": 0.0-1.0,
  "explanation": "简短说明",
  "extracted_info": {{"key": "value"}}
}}

意图类型说明：
- new_task: 用户想开始一个新任务（如"帮我订机票"）
- provide_info: 用户提供了追问中需要的信息
- confirm: 用户在确认/同意（如"好的"、"可以"）
- reject: 用户在拒绝（如"不对"、"不要这个"）
- modify: 用户想修改之前的指令
- cancel: 用户想取消当前任务
- question: 一般性问题，不涉及当前任务
- unclear: 意图不明确，需要进一步追问

只输出 JSON，不要其他文字。"""

        try:
            response = self.client.messages.create(
                model=self.model,
                max_tokens=300,
                messages=[{"role": "user", "content": prompt}],
                temperature=0.1,
            )
            text = response.content[0].text.strip()
            if "```" in text:
                text = text.split("```")[1].split("```")[0]
            if text.startswith("json"):
                text = text[4:]
            return json.loads(text)
        except Exception as e:
            # 降级：简单规则判断
            return self._fallback_recognize(user_input, current_state)

    def _fallback_recognize(self, user_input: str, state: DialogState) -> dict:
        """降级意图识别（基于规则）"""
        inp = user_input.lower().strip()

        if any(w in inp for w in ["取消", "算了", "不要了", "cancel"]):
            intent = "cancel"
        elif any(w in inp for w in ["好的", "可以", "没问题", "ok", "行"]):
            intent = "confirm"
        elif any(w in inp for w in ["不对", "不是", "改", "换成"]):
            intent = "modify"
        elif state.status != TaskStatus.IDLE:
            intent = "provide_info"  # 默认当作提供信息
        else:
            intent = "new_task"

        return {"intent": intent, "confidence": 0.6, "explanation": "规则降级", "extracted_info": {}}


# ================================================================
# 第3部分：动态规划器
# ================================================================

class DynamicPlanner:
    """动态调整任务计划"""

    def __init__(self, client: Anthropic, model: str = "claude-sonnet-4-6"):
        self.client = client
        self.model = model

    def create_plan(self, task: str, collected_info: dict) -> list[dict]:
        """为新任务创建执行计划"""
        prompt = f"""为以下任务创建分步执行计划。

任务: {task}
已知信息: {json.dumps(collected_info, ensure_ascii=False)}

请输出 JSON 数组格式：
[
  {{
    "step": 1,
    "action": "具体动作",
    "need_info": ["需要但还没有的信息"],
    "can_execute": true/false,
    "expected_output": "预期结果"
  }},
  ...
]

规则：
- 如果一个步骤需要的所有信息都已具备，can_execute 为 true
- 如果缺少关键信息，can_execute 为 false，并在 need_info 中列出
- 步骤要具体可执行
- 最后一步应该是"汇总并呈现结果"

只输出 JSON 数组。"""

        try:
            response = self.client.messages.create(
                model=self.model,
                max_tokens=800,
                messages=[{"role": "user", "content": prompt}],
                temperature=0.2,
            )
            text = response.content[0].text.strip()
            if "```" in text:
                text = text.split("```")[1].split("```")[0]
            return json.loads(text)
        except Exception:
            return [{"step": 1, "action": task, "need_info": [], "can_execute": True,
                     "expected_output": "完成任务"}]

    def adjust_plan(self, original_plan: list[dict], feedback: str, current_step: int) -> list[dict]:
        """根据用户反馈调整计划"""
        prompt = f"""用户对当前计划提出了修改意见。

原计划:
{json.dumps(original_plan, ensure_ascii=False, indent=2)}

当前步骤: 第{current_step}步
用户反馈: {feedback}

请输出调整后的完整计划（JSON 数组格式）。
只输出 JSON 数组。"""

        try:
            response = self.client.messages.create(
                model=self.model,
                max_tokens=800,
                messages=[{"role": "user", "content": prompt}],
                temperature=0.2,
            )
            text = response.content[0].text.strip()
            if "```" in text:
                text = text.split("```")[1].split("```")[0]
            return json.loads(text)
        except Exception:
            return original_plan


# ================================================================
# 第4部分：多轮对话 Agent
# ================================================================

class MultiTurnAgent:
    """多轮对话任务 Agent"""

    def __init__(self):
        self.client = Anthropic(api_key=os.getenv("ANTHROPIC_API_KEY"))
        self.model = "claude-sonnet-4-6"
        self.intent_recognizer = IntentRecognizer(self.client, self.model)
        self.planner = DynamicPlanner(self.client, self.model)
        self.state = DialogState()

    def handle(self, user_input: str) -> str:
        """主入口：处理用户输入，返回 Agent 回复"""
        print(f"\n{'─'*55}")
        print(f"👤 用户: {user_input}")

        # 步骤1：意图识别
        print("🎯 识别意图...")
        intent_result = self.intent_recognizer.recognize(user_input, self.state)
        intent = intent_result["intent"]
        print(f"  意图: {intent} (置信度: {intent_result.get('confidence', '?')})")

        # 步骤2：根据意图分发处理
        if intent == "cancel":
            response = self._handle_cancel()

        elif intent == "new_task":
            response = self._handle_new_task(user_input, intent_result)

        elif intent == "provide_info":
            response = self._handle_info(user_input, intent_result)

        elif intent == "confirm":
            response = self._handle_confirm()

        elif intent == "modify":
            response = self._handle_modify(user_input, intent_result)

        elif intent == "reject":
            response = self._handle_reject()

        elif intent == "question":
            response = self._handle_question(user_input)

        else:
            response = self._handle_unclear(user_input)

        # 保存对话历史
        self.state.add_turn(user_input, response)

        print(f"🤖 Agent: {response[:300]}...")
        return response

    # --- 意图处理函数 ---

    def _handle_new_task(self, user_input: str, intent: dict) -> str:
        """处理新任务"""
        # 如果有进行中的任务，先确认
        if self.state.status not in (TaskStatus.IDLE, TaskStatus.COMPLETED, TaskStatus.CANCELLED):
            return f"⚠️ 当前有进行中的任务「{self.state.task_description[:50]}...」。\n请先完成或取消当前任务，再开启新任务。\n输入'取消'来取消当前任务。"

        self.state.reset()
        self.state.status = TaskStatus.COLLECTING
        self.state.task_description = user_input
        self.state.user_intent = "new_task"

        # 提取初始信息
        extracted = intent.get("extracted_info", {})
        self.state.collected_info.update(extracted)

        # 用 Claude 分析还缺什么信息
        return self._analyze_missing_info(user_input)

    def _analyze_missing_info(self, task: str) -> str:
        """分析完成此任务还需要什么信息"""
        prompt = f"""用户想要: {task}
目前已知道的信息: {json.dumps(self.state.collected_info, ensure_ascii=False)}

请完成两件事：
1. 判断这个任务是什么类型（订票、写作、查询、分析、计算等）
2. 列出完成此任务还需要的信息（用友好的问题形式列出）

回复 JSON 格式：
{{
  "task_type": "任务类型",
  "questions": ["需要问的问题1", "需要问的问题2", ...],
  "can_proceed": true/false,
  "summary": "对任务的理解总结"
}}

如果信息已足够，can_proceed 设为 true，questions 为空数组。"""

        try:
            response = self.client.messages.create(
                model=self.model,
                max_tokens=500,
                messages=[{"role": "user", "content": prompt}],
                temperature=0.2,
            )
            text = response.content[0].text.strip()
            if "```" in text:
                text = text.split("```")[1].split("```")[0]
            if text.startswith("json"):
                text = text[4:]
            analysis = json.loads(text)
        except Exception:
            analysis = {"task_type": "通用任务", "questions": ["请详细描述你的需求？"],
                        "can_proceed": False, "summary": task}

        self.state.task_type = analysis.get("task_type", "通用任务")
        self.state.missing_info = analysis.get("questions", [])

        if analysis.get("can_proceed"):
            # 信息充足，直接开始规划
            self.state.status = TaskStatus.PLANNING
            return self._start_planning()

        # 信息不足，追问
        questions_text = "\n".join([f"{i+1}. {q}" for i, q in enumerate(self.state.missing_info)])
        return f"📋 我理解了，你想「{analysis.get('summary', task)}」。\n\n在开始之前，我需要了解更多：\n{questions_text}"

    def _handle_info(self, user_input: str, intent: dict) -> str:
        """处理用户提供的信息"""
        if self.state.status != TaskStatus.COLLECTING:
            # 不在收集状态，当作新的一般输入
            return self._handle_question(user_input)

        # 提取信息
        extracted = intent.get("extracted_info", {})
        self.state.collected_info.update(extracted)

        # 也把原始输入存起来
        info_key = f"info_{len(self.state.collected_info)}"
        self.state.collected_info[info_key] = user_input

        # 检查是否信息收集完毕
        if len(self.state.collected_info) >= 3 or self._info_sufficient():
            self.state.status = TaskStatus.PLANNING
            return self._start_planning()

        # 继续追问
        remaining = self.state.missing_info[len(self.state.collected_info):]
        if remaining:
            return f"收到！还需要了解一下：\n{remaining[0] if remaining else ''}"
        else:
            self.state.status = TaskStatus.PLANNING
            return self._start_planning()

    def _info_sufficient(self) -> bool:
        """判断已收集的信息是否足够"""
        # 简单判断：至少收集了3条信息 + 没有明显缺失
        if len(self.state.collected_info) < 2:
            return False
        # 或者所有 missing_info 都有对应的收集了
        return len(self.state.collected_info) >= len(self.state.missing_info)

    def _start_planning(self) -> str:
        """开始规划"""
        print("📋 规划中...")
        plan = self.planner.create_plan(
            self.state.task_description,
            self.state.collected_info
        )
        self.state.plan = plan
        self.state.status = TaskStatus.WAITING_FEEDBACK

        # 展示计划给用户确认
        plan_text = "\n".join([
            f"  步骤{s['step']}: {s['action']} {'✅可执行' if s.get('can_execute') else '⚠️待补充信息'}"
            for s in plan
        ])

        return f"📋 我制定了以下执行计划：\n\n{plan_text}\n\n确认执行吗？（回复'好的'开始，或提出修改意见）"

    def _handle_confirm(self) -> str:
        """处理用户确认"""
        if self.state.status == TaskStatus.WAITING_FEEDBACK:
            self.state.status = TaskStatus.EXECUTING
            return self._execute_plan()
        return "好的！请问还有什么需要帮助的吗？"

    def _handle_modify(self, user_input: str, intent: dict) -> str:
        """处理用户修改"""
        if self.state.plan:
            self.state.plan = self.planner.adjust_plan(
                self.state.plan, user_input, self.state.current_step
            )
            self.state.status = TaskStatus.WAITING_FEEDBACK

            plan_text = "\n".join([
                f"  步骤{s['step']}: {s['action']}"
                for s in self.state.plan
            ])
            return f"🔄 已按你的意见调整计划：\n\n{plan_text}\n\n确认执行吗？"

        return "好的，那你具体想怎么调整呢？"

    def _handle_reject(self) -> str:
        """处理用户拒绝"""
        self.state.reset()
        return "好的，已取消当前计划。请告诉我你想做什么？"

    def _handle_cancel(self) -> str:
        """处理取消"""
        task = self.state.task_description[:50]
        self.state.reset()
        self.state.status = TaskStatus.CANCELLED
        return f"✅ 已取消「{task}...」。有什么可以帮你的吗？"

    def _handle_question(self, user_input: str) -> str:
        """处理一般性问题（无任务上下文）"""
        try:
            response = self.client.messages.create(
                model=self.model,
                max_tokens=500,
                messages=[{"role": "user", "content": user_input}],
                temperature=0.3,
            )
            return response.content[0].text.strip()
        except Exception as e:
            return f"抱歉，出了点问题: {e}"

    def _handle_unclear(self, user_input: str) -> str:
        """处理意图不明确的输入"""
        return f"🤔 我不太确定你的意思。你是想：\n1. 开始一个新任务\n2. 提供之前任务的补充信息\n3. 问一个一般性问题\n\n能详细说说吗？"

    def _execute_plan(self) -> str:
        """执行计划（简化版：用 Claude 模拟执行）"""
        if not self.state.plan:
            return "⚠️ 没有可执行的计划"

        results = []
        for step in self.state.plan:
            if not step.get("can_execute", False):
                results.append(f"⚠️ 步骤{step['step']}信息不足，跳过: {step['action']}")
                continue

            self.state.current_step = step["step"]
            print(f"  ⚡ 执行步骤{step['step']}: {step['action']}")

            # 用 Claude 执行这一步
            try:
                response = self.client.messages.create(
                    model=self.model,
                    max_tokens=400,
                    messages=[{"role": "user", "content":
                        f"执行以下任务步骤（请给出具体结果，简洁回答，不需要解释过程）：\n\n"
                        f"任务: {self.state.task_description}\n"
                        f"已有信息: {json.dumps(self.state.collected_info, ensure_ascii=False)}\n"
                        f"当前步骤: {step['action']}\n"
                        f"预期输出: {step.get('expected_output', '')}"}],
                    temperature=0.3,
                )
                step_result = response.content[0].text.strip()
                results.append(f"✅ 步骤{step['step']}: {step_result[:150]}")
            except Exception as e:
                results.append(f"❌ 步骤{step['step']}失败: {e}")

        self.state.execution_results = results
        self.state.status = TaskStatus.COMPLETED

        summary = "\n".join(results)
        final = f"✅ 任务完成！\n\n{summary}\n\n还有什么需要帮助的吗？"

        # 重置任务状态（但保留对话历史）
        task = self.state.task_description
        self.state.status = TaskStatus.COMPLETED
        return final


# ================================================================
# 第5部分：交互入口
# ================================================================

def main():
    print("=" * 55)
    print("💬 MultiTurnAgent — 多轮对话任务 Agent")
    print("=" * 55)
    print("支持多轮交互，试试这些场景：")
    print("  场景1: 帮我订一张机票")
    print("  场景2: 帮我写一份项目周报")
    print("  场景3: 帮我做一个旅行计划")
    print("  场景4: 北京今天天气怎么样？（一般问答）")
    print("命令: status(查看状态) | quit(退出)")
    print("=" * 55)

    if not os.getenv("ANTHROPIC_API_KEY"):
        print("❌ 请先配置 ANTHROPIC_API_KEY！")
        return

    agent = MultiTurnAgent()

    print("\n🤖 Agent: 你好！我是你的智能助手。请告诉我你需要什么帮助？")

    while True:
        try:
            user_input = input("\n👤 你: ").strip()
            if not user_input:
                continue
            if user_input.lower() in ("quit", "exit", "q"):
                break
            if user_input.lower() == "status":
                print(f"📊 当前状态: {agent.state.summary()}")
                continue

            response = agent.handle(user_input)
            print(f"\n🤖 Agent:\n{response}")

        except KeyboardInterrupt:
            break

    print(f"\n👋 MultiTurnAgent 已退出。再见！")


if __name__ == "__main__":
    main()

4. 状态流转图

5. 调试与常见问题