多轮对话：正确传递工具调用历史

curl -X POST 'https://dashscope.aliyuncs.com/api/v2/apps/knowledge/chat' \
  -H 'Authorization: Bearer $API_KEY' \
  -H 'Content-Type: application/json' \
  -d '{
    "input": {
      "messages": [{"role": "user", "content": "如何调用通义千问 API"}]
    },
    "parameters": {
      "agent_options": {"agent_id": "aid-xxxxxxxx"}
    },
    "stream": true
  }'

curl -X POST 'https://dashscope.aliyuncs.com/api/v2/apps/knowledge/chat' \
  -H 'Authorization: Bearer $API_KEY' \
  -H 'Content-Type: application/json' \
  -d '{
    "input": {
      "messages": [
        {"role": "user", "content": "如何调用通义千问 API"},
        {"role": "assistant", "content": "", "tool_calls": [{"id":"call_956...","function":{"name":"semantic_search",...}}]},
        {"role": "tool", "tool_call_id": "call_956...", "content": "检索到 5 条相关切片..."},
        {"role": "assistant", "content": "根据检索到的《首次调用千问API》内容..."},
        {"role": "user", "content": "如何把 API Key 配置到环境变量？"}
      ]
    },
    "parameters": {"agent_options": {"agent_id": "aid-xxxxxxxx"}},
    "stream": true
  }'

import json
import requests

# 从控制台问答服务的 API 调试页面获取
API_URL = "https://dashscope.aliyuncs.com/api/v2/apps/knowledge/chat"
API_KEY = "sk-xxxxxxxx"      # workspace-scoped API Key（sk-ws- 前缀）
AGENT_ID = "aid-xxxxxxxx"    # 知识问答服务 ID

class ConversationTracker:
    """追踪单轮 Agent 响应，从流式事件提取完整消息序列。"""

    def __init__(self):
        self.messages = []          # 提取出的消息序列（工具调用 + 工具返回）
        self.tool_call_rounds = 0   # 工具调用轮数（用于效果对比）
        self.answer = ""            # 最终回答
        self._generating = False

    def process_sse_line(self, line: str):
        if not line.startswith("data:"):
            return
        obj = json.loads(line[len("data:"):].strip())
        choices = obj.get("output", {}).get("choices", [])
        if not choices:
            return

        msg = choices[0].get("message", {})
        step_change = msg.get("extra", {}).get("step_change", "")
        role = msg.get("role", "")
        content = msg.get("content", "") or ""

        # 事件 1：工具调用派发（step_change == "tool_calling"）→ 提取 tool_calls
        if step_change == "tool_calling" and msg.get("tool_calls"):
            self.tool_call_rounds += 1
            self.messages.append({
                "role": "assistant", "content": "",
                "tool_calls": [{"id": tc["id"], "type": "function",
                                "function": {"name": tc["function"]["name"],
                                             "arguments": tc["function"]["arguments"]}}
                               for tc in msg["tool_calls"]],
            })
        # 事件 2：工具返回（step_change == "tool_return"，role == "tool"）→ 提取 content + tool_call_id
        elif step_change == "tool_return" and role == "tool":
            self.messages.append({"role": "tool",
                                  "tool_call_id": msg.get("tool_call_id", ""),
                                  "content": content})
        # 事件 3：最终回答（generation_start → 内容流 → generation_end）→ 累积 content
        elif step_change == "generation_start":
            self._generating = True
            self.answer = content
        elif self._generating and content:
            self.answer += content
        elif step_change == "generation_end":
            self.answer += content
            self._generating = False

    def get_response_messages(self):
        """返回本轮提取的完整消息序列（含最终回答）。"""
        result = list(self.messages)
        if self.answer:
            result.append({"role": "assistant", "content": self.answer})
        return result

class MultiTurnChat:
    """管理多轮对话历史，自动携带工具调用链路。"""

    def __init__(self):
        self.history = []

    def ask(self, question: str) -> str:
        """发送问题，自动拼接完整历史（含工具调用链路），返回最终回答。"""
        self.history.append({"role": "user", "content": question})
        sent = len(self.history)
        tracker = self._call_agent(self.history)
        # 将本轮提取的工具调用链路和最终回答加入历史
        self.history.extend(tracker.get_response_messages())
        print(f"  发送 {sent} 条消息，工具调用 {tracker.tool_call_rounds} 轮")
        return tracker.answer

    def _call_agent(self, messages):
        headers = {"Authorization": f"Bearer {API_KEY}", "Content-Type": "application/json"}
        body = {
            "input": {"messages": messages},
            "parameters": {"agent_options": {"agent_id": AGENT_ID}},
            "stream": True,
        }
        tracker = ConversationTracker()
        resp = requests.post(API_URL, headers=headers, json=body, stream=True, timeout=180)
        for chunk in resp.iter_lines():
            if chunk:
                tracker.process_sse_line(chunk.decode("utf-8").strip())
        return tracker

chat = MultiTurnChat()

# 第 1 轮
print("[第 1 轮] 如何调用通义千问 API")
chat.ask("如何调用通义千问 API")

# 第 2 轮 · 传完整工具历史（自动携带）
print("\n[第 2 轮 · 传完整工具历史]")
chat_full = MultiTurnChat()
chat_full.history = list(chat.history)
chat_full.ask("如何把 API Key 配置到环境变量？")

# 第 2 轮 · 只传文本历史（丢弃工具调用，只留 user + assistant 回答）
print("\n[第 2 轮 · 只传文本历史]")
text_only = [m for m in chat.history
             if m["role"] == "user" or (m["role"] == "assistant" and not m.get("tool_calls"))]
chat_text = MultiTurnChat()
chat_text.history = text_only
chat_text.ask("如何把 API Key 配置到环境变量？")

[第 1 轮] 如何调用通义千问 API
  发送 1 条消息，工具调用 2 轮

[第 2 轮 · 传完整工具历史]
  发送 7 条消息，工具调用 0 轮      ← 直接复用，未重复检索

[第 2 轮 · 只传文本历史]
  发送 3 条消息，工具调用 2 轮      ← 重新检索

def truncate_history(messages, max_tool_len=2000, keep_recent=2):
    """截断较早轮次的工具返回内容。"""
    user_indices = [i for i, m in enumerate(messages) if m["role"] == "user"]
    if len(user_indices) <= keep_recent:
        return messages
    cutoff = user_indices[-keep_recent]
    result = []
    for i, msg in enumerate(messages):
        if i < cutoff and msg["role"] == "tool" and len(msg["content"]) > max_tool_len:
            msg = {**msg, "content": msg["content"][:max_tool_len] + "\n...(已截断)"}
        result.append(msg)
    return result