1
# ---------- 2-1 自动摘要 ----------
2
import os
3

4
from dotenv import load_dotenv
5
from langchain.agents import create_agent
6
from langchain.agents.middleware import SummarizationMiddleware
7
from langchain.chat_models import init_chat_model
8
from langchain.messages import HumanMessage
9

10
load_dotenv(override=True)
11

12
model = init_chat_model(
13
    model="deepseek-v4-flash",
14
    model_provider="openai",
15
    api_key=os.getenv("DEEPSEEK_API_KEY"),
16
    base_url=os.getenv("DEEPSEEK_BASE_URL"),
17
)
18

19
agent_sum = create_agent(
20
    model=model,
21
    tools=[],
22
    middleware=[
23
        SummarizationMiddleware(
24
            model=model,                    # 摘要用哪个模型
25
            trigger=[("messages", 6)],      # 消息数达到 6 条就触发
26
            keep=("messages", 2),           # 摘要时保留最近 2 条原文
27
        )
28
    ],
29
)
30

31
messages = [HumanMessage("你好，我是老王")]
32
for i in range(4):
33
    r = agent_sum.invoke({"messages": messages})
34
    messages = r["messages"]
35
    print(f"第 {i + 1} 轮: 共 {len(messages)} 条 | {[type(m).__name__ for m in messages]}")
36
    messages.append(HumanMessage(f"继续聊，这是第 {i + 2} 句"))
37
# 触发那一轮的条数会明显掉下来，第一条变成内容形如
38
# "Here is a summary of the conversation to date: ..." 的 HumanMessage
39

40
# ---------- 2-1 挑战：fraction 触发要先有 profile ----------
41
try:
42
    agent_frac = create_agent(
43
        model=model,
44
        tools=[],
45
        middleware=[SummarizationMiddleware(model=model, trigger=[("fraction", 0.5)])],
46
    )
47
    agent_frac.invoke({"messages": [HumanMessage("你好")]})
48
except ValueError as e:
49
    print("意料之中的报错:", str(e)[:60], "...")
50

51
# 解法：手动补上 profile（DeepSeek-V3.2 的上下文长度是 128K）
52
from langchain_deepseek import ChatDeepSeek
53

54
model_p = ChatDeepSeek(
55
    model="deepseek-v4-flash",
56
    api_key=os.getenv("DEEPSEEK_API_KEY"),
57
    api_base=os.getenv("DEEPSEEK_BASE_URL"),
58
    profile={"max_input_tokens": 128_000},
59
)
60
agent_frac_ok = create_agent(
61
    model=model_p,
62
    tools=[],
63
    middleware=[SummarizationMiddleware(model=model_p, trigger=[("fraction", 0.5)])],
64
)
65
print("补了 profile 之后:", agent_frac_ok.invoke({"messages": [HumanMessage("你好")]})["messages"][-1].content)
66
# ---------- 2-2 PII 的四种处理策略 ----------
67
from langchain.agents.middleware import PIIMiddleware
68

69
def ask(strategy: str, text: str, pii_type: str = "email"):
70
    """用指定策略建 Agent，问一句带敏感信息的话，返回模型回答"""
71
    agent = create_agent(
72
        model=model,
73
        tools=[],
74
        middleware=[PIIMiddleware(pii_type, strategy=strategy, apply_to_input=True)],
75
    )
76
    return agent.invoke({"messages": [HumanMessage(text)]})["messages"][-1].content
77

78
# 让模型"原样复述"最容易看出脱敏效果：它只能复述自己真正看到的内容
79
print("原文     :", "我的邮箱是 zhangsan@example.com")
80
for s in ["redact", "mask", "hash"]:
81
    print(f"{s:8s} :", ask(s, "我的邮箱是 zhangsan@example.com。请原样复述我这句话。"))
82

83
# 银行卡号用 mask 看得最清楚（注意：内置检测器要求卡号能通过 Luhn 校验，"…1234" 这种编造号认不出来）
84
print("信用卡 mask:", ask("mask", "我的卡号是 4111-1111-1111-1111。请原样复述我这句话。", "credit_card"))
85

86
# block：检测到就直接抛异常，Agent 根本不会启动模型调用
87
try:
88
    ask("block", "我的邮箱是 zhangsan@example.com")
89
except Exception as e:
90
    print("block    :", type(e).__name__, str(e)[:80])
91
# ---------- 2-3 关键工具调用前先审批 ----------
92
from langchain.agents.middleware import HumanInTheLoopMiddleware
93
from langchain.tools import tool
94
from langgraph.checkpoint.memory import InMemorySaver
95
from langgraph.types import Command
96

97
@tool
98
def send_email_tool(recipient: str, subject: str, body: str) -> str:
99
    """发送邮件
100

101
    Args:
102
        recipient: 收件人邮箱
103
        subject: 邮件标题
104
        body: 邮件正文
105
    """
106
    print("   >>> 真的执行发送邮件工具了")
107
    return f"已发送给 {recipient}：{subject}"
108

109
agent_hitl = create_agent(
110
    model=model,
111
    tools=[send_email_tool],
112
    checkpointer=InMemorySaver(),            # 硬性要求①：没有它中断后没法"接着跑"
113
    middleware=[
114
        HumanInTheLoopMiddleware(
115
            interrupt_on={
116
                "send_email_tool": {
117
                    "allowed_decisions": ["approve", "edit", "reject"],   # 三种决策都允许
118
                    "description": "要发邮件了，请确认",
119
                },
120
            },
121
            description_prefix="需要你审批：",
122
        ),
123
    ],
124
)
125

126
QUESTION = "帮我给 abc@qq.com 发一封邮件，标题'测试'，内容'你好'"
127

128
def run(decision: dict, thread_id: str, label: str):
129
    config = {"configurable": {"thread_id": thread_id}}     # 硬性要求②：恢复要用同一个 config
130
    resp = agent_hitl.invoke({"messages": [HumanMessage(QUESTION)]}, config=config)
131
    interrupts = resp.get("__interrupt__", [])
132
    print(f"[{label}] 中断了吗: {bool(interrupts)}")
133
    for req in interrupts[0].value["action_requests"]:
134
        print(f"   待审批: {req['name']} | 参数 {req['args']}")
135
    out = agent_hitl.invoke(Command(resume={"decisions": [decision]}), config=config)
136
    for m in out["messages"]:
137
        if type(m).__name__ == "ToolMessage":
138
            print(f"   工具结果: {str(m.content)[:70]}")
139

140
run({"type": "reject"}, "1", "reject 拒绝")
141
run({"type": "approve"}, "2", "approve 同意")
142
run(
143
    {
144
        "type": "edit",
145
        "edited_action": {
146
            "name": "send_email_tool",
147
            "args": {"recipient": "abc@qq.com", "subject": "改过的标题", "body": "改过的正文"},
148
        },
149
    },
150
    "3",
151
    "edit 改参数",
152
)
153
# ---------- 2-4 待办清单：先列清单再动手 ----------
154
from langchain.agents.middleware import TodoListMiddleware
155
from langchain.tools import tool
156

157
@tool
158
def add(a: int, b: int) -> int:
159
    """返回两个整数的和
160

161
    Args:
162
        a: 第一个整数
163
        b: 第二个整数
164
    """
165
    return a + b
166

167
@tool
168
def multiply(a: int, b: int) -> int:
169
    """返回两个整数的积
170

171
    Args:
172
        a: 第一个整数
173
        b: 第二个整数
174
    """
175
    return a * b
176

177
# 挂上中间件：模型能用的工具里会自动多一个"写清单"的工具
178
agent_todo = create_agent(
179
    model=model,
180
    tools=[add, multiply],
181
    middleware=[TodoListMiddleware()],
182
    system_prompt="遇到多步骤任务时，先用 write_todos 列出计划，再一步一步执行。",
183
)
184

185
TASK = "先算 12+34，再算这个结果的 3 倍，最后把两步的结果加起来是多少？"
186
final_state = agent_todo.invoke({"messages": [HumanMessage(TASK)]})
187

188
# ① 这次运行一共更新了几次清单，每次清单里每条事项的状态
189
updates = []
190
for m in final_state["messages"]:
191
    for tc in getattr(m, "tool_calls", None) or []:
192
        if tc["name"] == "write_todos":
193
            updates.append(tc["args"]["todos"])
194
    if type(m).__name__ == "ToolMessage" and m.name == "write_todos":
195
        print("工具返回:", str(m.content)[:56], "...")
196
print(f"一共更新了 {len(updates)} 次清单：")
197
for i, todos in enumerate(updates, 1):
198
    print(f"  第 {i} 次:", [(t["content"], t["status"]) for t in todos])
199

200
# ② state 里那份独立于消息列表的清单（前端进度条读的就是它）
201
print("最终清单（来自 state，不属于 messages）:")
202
for i, item in enumerate(final_state.get("todos", []), 1):
203
    print(f"  {i}. [{item.get('status')}] {item.get('content')}")
204

205
# ③ 这次运行里出现过的三种状态
206
seen = {t["status"] for todos in updates for t in todos}
207
print("出现过的状态:", sorted(seen))
208
print("最终回答:", final_state["messages"][-1].content)
209

210
# 挑战：同一个任务，不挂中间件再跑一次（请求里没有"写清单"这个工具）
211
agent_plain = create_agent(model=model, tools=[add, multiply])
212
plain_state = agent_plain.invoke({"messages": [HumanMessage(TASK)]})
213
plain_calls = [tc["name"] for m in plain_state["messages"] for tc in (getattr(m, "tool_calls", None) or [])]
214
print("不挂中间件时的工具调用:", plain_calls, "| state 里有 todos 吗:",
215
      "todos" in plain_state)

1
# ---------- 3-1 综合题：让 Agent 自己"扫描 → 跑测试 → 改代码 → 再跑测试" ----------
2
import os
3
import subprocess
4
import sys
5
from pathlib import Path
6

7
from dotenv import load_dotenv
8
from langchain.agents import create_agent
9
from langchain.agents.middleware import TodoListMiddleware
10
from langchain.chat_models import init_chat_model
11
from langchain.messages import HumanMessage
12
from langchain.tools import tool
13

14
load_dotenv(override=True)
15

16
model = init_chat_model(
17
    model="deepseek-v4-flash",
18
    model_provider="openai",
19
    api_key=os.getenv("DEEPSEEK_API_KEY"),
20
    base_url=os.getenv("DEEPSEEK_BASE_URL"),
21
)
22

23
# 第一步：准备一个"有 Bug"的工作区（my_add.py 里故意把和写成差）
24
WORKSPACE = Path(__file__).resolve().parent / "todo_workspace"
25
WORKSPACE.mkdir(exist_ok=True)
26
(WORKSPACE / "my_add.py").write_text(
27
    'def add(a: int, b: int) -> int:\n    """返回两个整数的和"""\n    return a - b\n',
28
    encoding="utf-8",
29
)
30
(WORKSPACE / "test_my_add.py").write_text(
31
    "from my_add import add\n\n\ndef test_add():\n"
32
    '    """测试加法功能"""\n'
33
    "    assert add(2, 3) == 5\n"
34
    "    assert add(-1, 1) == 0\n"
35
    "    assert add(0, 0) == 0\n"
36
    "    assert add(10, -5) == 5\n",
37
    encoding="utf-8",
38
)
39

40
# 第二步：四个工具（都先 resolve 再判断前缀，防止模型用 ../../ 跑出工作区）
41
@tool
42
def list_files(path: str = ".") -> str:
43
    """列出工作区指定目录下的文件和子目录。path 只能是相对路径。
44

45
    Args:
46
        path: 工作区下的相对路径，一定指向目录，默认为 .（工作区根目录）
47
    """
48
    target = (WORKSPACE / path).resolve()
49
    root = WORKSPACE.resolve()
50
    if not str(target).startswith(str(root)):
51
        return "错误：只允许访问工作区内的目录。"
52
    if not target.is_dir():
53
        return f"错误：不是目录: {path}"
54
    lines = [
55
        f"[{'DIR' if p.is_dir() else 'FILE'}] {p.relative_to(root).as_posix()}"
56
        for p in sorted(target.iterdir(), key=lambda p: (p.is_file(), p.name.lower()))
57
    ]
58
    return "\n".join(lines) or f"目录为空: {path}"
59

60
@tool
61
def read_file(path: str) -> str:
62
    """读取工作区中的文本文件内容。path 只能是相对路径。
63

64
    Args:
65
        path: 工作区内的文件名
66
    """
67
    file_path = (WORKSPACE / path).resolve()
68
    if not str(file_path).startswith(str(WORKSPACE.resolve())):
69
        return "错误：只允许读取工作区内的文件。"
70
    if not file_path.exists():
71
        return f"错误：文件不存在: {path}"
72
    return file_path.read_text(encoding="utf-8")
73

74
@tool
75
def write_file(path: str, content: str) -> str:
76
    """写入工作区中的文本文件。path 只能是相对路径。
77

78
    Args:
79
        path: 工作区内的文件名
80
        content: 写入文件的内容
81
    """
82
    file_path = (WORKSPACE / path).resolve()
83
    if not str(file_path).startswith(str(WORKSPACE.resolve())):
84
        return "错误：只允许写入工作区内的文件。"
85
    file_path.write_text(content, encoding="utf-8")
86
    return f"已写入文件: {path}"
87

88
@tool
89
def run_tests() -> str:
90
    """在工作区运行 pytest -q 并返回输出。不接收任何参数。"""
91
    try:
92
        result = subprocess.run(
93
            # 用"当前解释器 -m pytest"，就不会踩"环境里没有 pytest 命令"的坑
94
            [sys.executable, "-m", "pytest", "-q"],
95
            cwd=str(WORKSPACE),
96
            capture_output=True,
97
            text=True,
98
            timeout=60,
99
        )
100
        return (
101
            f"returncode={result.returncode}\n\n"
102
            f"STDOUT:\n{result.stdout}\n\n"
103
            f"STDERR:\n{result.stderr}"
104
        )
105
    except Exception as e:  # noqa: BLE001
106
        return f"运行测试失败: {e}"
107

108
# 第三步：挂上待办清单中间件跑起来
109
agent = create_agent(
110
    model=model,
111
    tools=[list_files, read_file, write_file, run_tests],   # 第 5 个「写清单」工具由中间件自动加
112
    middleware=[TodoListMiddleware()],
113
    system_prompt=(
114
        "你是一个代码修复助手。遇到多步骤任务时，先使用 write_todos 制定待办事项；"
115
        "然后读取文件、修复代码并运行测试。工作全部在工作区下进行。"
116
    ),
117
)
118

119
print("正在执行 Agent 任务...")
120
final_state = agent.invoke(
121
    {"messages": [HumanMessage(content="请测试并修复工作区下 my_add.py 文件中的代码")]}
122
)
123

124
# 第四步：从消息里找出模型写的清单（write_todos 的调用与工具返回）
125
for m in final_state["messages"]:
126
    for tc in getattr(m, "tool_calls", None) or []:
127
        if tc["name"] == "write_todos":
128
            print("模型写了清单:", [t["content"] for t in tc["args"]["todos"]])
129
    if type(m).__name__ == "ToolMessage" and m.name == "write_todos":
130
        print("工具返回:", str(m.content)[:60], "...")
131
    if type(m).__name__ == "ToolMessage" and m.name == "run_tests":
132
        print("测试结果:", str(m.content).splitlines()[0])      # returncode=0 就是全绿
133

134
# todos 独立于 messages，存在 state 里 —— 前端进度条读它就行
135
for i, item in enumerate(final_state.get("todos", []), 1):
136
    print(f"{i}. [{item.get('status')}] {item.get('content')}")
137

138
print("修复后的文件内容:", (WORKSPACE / "my_add.py").read_text(encoding="utf-8").splitlines()[-1])