1
 2
 3
 4
 5
 6
 7
 8
 9
10
11

def trim_messages(messages, keep_recent=10):
    """滑窗：永远保留 system，只留最近 keep_recent 条消息"""
    system = messages[0]  # system 提示永远不丢
    recent = messages[-keep_recent:]

    # 注意：不要让窗口以一条孤立的 tool result 开头
    # tool result 前面必须跟着它对应的 tool_call，否则模型报错
    while recent and recent[0]["role"] == "tool":
        recent = recent[1:]

    return [system] + recent

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29

def summarize_old_messages(messages, keep_recent=6):
    """把老消息压缩成一段摘要，保留最近 keep_recent 条原文"""
    if len(messages) <= keep_recent + 1:
        return messages

    system = messages[0]
    old = messages[1:-keep_recent]  # 要被压缩的老消息
    recent = messages[-keep_recent:]  # 最近的几轮保留原文

    # 把老消息拼成文本，让模型总结
    old_text = "\n".join([
        f"[{m['role']}] {m.get('content', '') or '(tool_call)'}"
        for m in old
    ])

    resp = client.chat.completions.create(
        model="deepseek-v4-flash",
        messages=[
            {"role": "system", "content": "你是对话摘要助手。保留关键事实、用户偏好和已确认的结论，不要遗漏数据。"},
            {"role": "user", "content": f"把以下对话压缩成一段简短摘要：\n\n{old_text}"}
        ]
    )
    summary = resp.choices[0].message.content

    # 新的 messages = system + 摘要 + 最近的原文
    return [
        system,
        {"role": "user", "content": f"[之前的对话摘要] {summary}"},
    ] + recent

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84

import json
from openai import OpenAI

client = OpenAI(base_url="https://api.deepseek.com", api_key="你的DeepSeek API Key")  # 去 platform.deepseek.com 注册获取

# --- 工具函数（和上一篇一样）---
def get_weather(city: str) -> str:
    db = {"北京": "晴 28°C", "上海": "多云 25°C", "深圳": "雷阵雨 31°C"}
    return db.get(city, f"暂无{city}的天气数据")

def get_aqi(city: str) -> str:
    db = {"北京": "85（轻度污染）", "上海": "42（优）", "深圳": "30（优）"}
    return db.get(city, f"暂无{city}的 AQI 数据")

tools = [
    {"type": "function", "function": {"name": "get_weather", "description": "查天气", "parameters": {"type": "object", "properties": {"city": {"type": "string"}}, "required": ["city"]}}},
    {"type": "function", "function": {"name": "get_aqi", "description": "查空气质量", "parameters": {"type": "object", "properties": {"city": {"type": "string"}}, "required": ["city"]}}},
]
available_functions = {"get_weather": get_weather, "get_aqi": get_aqi}

MAX_TURNS = 5
MAX_HISTORY = 20  # 超过这个条数就触发摘要

def summarize_old_messages(messages, keep_recent=6):
    if len(messages) <= keep_recent + 1:
        return messages
    system = messages[0]
    old = messages[1:-keep_recent]
    recent = messages[-keep_recent:]
    old_text = "\n".join([f"[{m['role']}] {m.get('content', '') or '(tool_call)'}" for m in old])
    resp = client.chat.completions.create(
        model="deepseek-v4-flash",
        messages=[
            {"role": "system", "content": "你是对话摘要助手。保留关键事实和用户偏好。"},
            {"role": "user", "content": f"压缩成简短摘要：\n\n{old_text}"}
        ]
    )
    summary = resp.choices[0].message.content
    return [system, {"role": "user", "content": f"[之前的对话摘要] {summary}"}] + recent

messages = [
    {"role": "system", "content": "你是一个帮用户做决策的助手。需要数据时自己调工具，信息够了就直接给结论。"}
]

print("开始聊天吧，输入 exit 退出。")
while True:
    user_input = input("\n你: ").strip()
    if user_input.lower() == "exit":
        break
    if not user_input:
        continue

    messages.append({"role": "user", "content": user_input})

    # Agent 内部工具循环（和上一篇一模一样）
    for turn in range(MAX_TURNS):
        resp = client.chat.completions.create(
            model="deepseek-v4-flash",
            messages=messages,
            tools=tools,
        )
        msg = resp.choices[0].message
        messages.append(msg.model_dump(exclude_none=True))

        if not msg.tool_calls:
            print(f"\n助手: {msg.content}")
            break

        for call in msg.tool_calls:
            fn_name = call.function.name
            fn_args = json.loads(call.function.arguments)
            print(f"  [tool] {fn_name}({fn_args})")
            try:
                result = available_functions[fn_name](**fn_args)
            except Exception as e:
                result = f"工具执行失败：{type(e).__name__}: {e}"
            messages.append({"role": "tool", "tool_call_id": call.id, "content": result})
    else:
        print("（达到最大轮次，强制停止）")

    # 对话太长了，压缩一下
    if len(messages) > MAX_HISTORY:
        print("  [记忆] 对话太长，压缩中...")
        messages = summarize_old_messages(messages, keep_recent=6)