1
from langchain.chat_models import init_chat_model
2
from dotenv import load_dotenv
3
import os
4

5
# 从.env文件中加载环境变量
6
load_dotenv(override=True)
7

8
custom_profile = {
9
    "max_input_tokens": 128_000
10
}
11

12
model = init_chat_model(
13
    model="gpt-5.4-mini",
14
    model_provider="openai",
15
    profile=custom_profile,
16
    api_key=os.getenv("CLOSEAI_API_KEY"),
17
    base_url=os.getenv("CLOSEAI_BASE_URL")
18
)

1
from langchain.chat_models import init_chat_model
2
from dotenv import load_dotenv
3
import os
4

5
# 从.env文件中加载环境变量
6
load_dotenv(override=True)
7

8
custom_profile = {
9
    "max_input_tokens": 128_000
10
}
11

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

1
from langchain.agents import create_agent
2
from langchain.agents.middleware import SummarizationMiddleware
3
from langchain.messages import SystemMessage, HumanMessage, AIMessage
4

5
messages = [
6
    SystemMessage("你是个非常友好的AI助手"),
7
    HumanMessage("你好啊，我是老王，你是谁？"),
8
    AIMessage("你好老王，我是小王"),
9
    HumanMessage("好的小王，很高兴认识你"),
10
    AIMessage("你高兴得太早了"),
11
    HumanMessage("呵呵，你什么意思")
12
]
13

14
agent = create_agent(
15
    model="deepseek-v4-flash",
16
    middleware=[
17
        SummarizationMiddleware(
18
            model=model,
19
            trigger=[
20
                ("tokens", 100),
21
                ("messages", 6),
22
                ("fraction", 0.001)
23
            ],
24
            keep=("messages", 2)
25
        )
26
    ]
27
)
28

29
response = agent.invoke({
30
    "messages": messages
31
})
32

33
for msg in response["messages"]:
34
    msg.pretty_print()

1
================================ Human Message ================================
2
Here is a summary of the conversation to date:
3
## SESSION INTENT
4
用户（老王）与AI（小王）进行初次问候和介绍。没有明确的后续任务目标，会话目前处于社交开场阶段。
5
## SUMMARY
6
用户自称"老王"，向AI问好并询问AI的身份。AI回应，自我介绍为"小王"。用户随后表示很高兴认识AI。这是一段简短的社交性对话开端，没有涉及具体任务、决策或深入讨论。
7
## ARTIFACTS
8
None
9
## NEXT STEPS
10
等待用户提出具体的请求或任务。如果会话继续，可能需要明确用户的意图或需求。
11
================================== Ai Message ==================================
12
你高兴得太早了
13
================================= Human Message =================================
14
呵呵，你什么意思
15
================================== Ai Message ==================================
16
"你高兴得太早了"是一句常见的网络调侃用语，字面意思是"别急着开心，后面可能有变化或挑战"。我作为AI没有真实情绪，但如果你在测试我的反应模式——我理解这是一种幽默或反讽的表达方式。
17
需要我帮忙处理具体事务时，可以随时告诉我哦

1
from langchain.agents import create_agent
2
from langchain.agents.middleware import SummarizationMiddleware
3
from langchain.messages import SystemMessage, HumanMessage, AIMessage
4
from langchain.chat_models import init_chat_model
5
from dotenv import load_dotenv
6
import os
7

8
# 从.env文件中加载环境变量
9
load_dotenv(override=True)
10

11
custom_profile = {
12
    "max_input_tokens": 1_000_000
13
}
14

15
model = init_chat_model(
16
    model="gpt-5.4-mini",
17
    model_provider="openai",
18
    profile=custom_profile,
19
    api_key=os.getenv("CLOSEAI_API_KEY"),
20
    base_url=os.getenv("CLOSEAI_BASE_URL")
21
)
22

23
messages = [
24
    SystemMessage("你是个非常有好的AI助手"),
25
    HumanMessage("你好啊，我是老王，你是谁？"),
26
    AIMessage("你好老王，我是小王"),
27
    HumanMessage("好的小王，很高兴认识你"),
28
    AIMessage("你高兴得太早了"),
29
    HumanMessage("呵呵，你什么意思，你是谁？")
30
]
31

32
agent = create_agent(
33
    model="deepseek-v4-flash",
34
    middleware=[
35
        SummarizationMiddleware(
36
            model=model,
37
            trigger=[
38
                ("tokens", 100),
39
                ("messages", 6),
40
                ("fraction", 0.0001)
41
            ],
42
            keep=("messages", 2),
43
            summary_prompt="对历史消息摘要，消息列表如下\n{messages}"
44
        )
45
    ]
46
)
47

48
response = agent.invoke({
49
    "messages": messages
50
})
51

52
for msg in response["messages"]:
53
    msg.pretty_print()

1
================================= Human Message =================================
2
对历史消息的摘要如下：
3
用户先自我介绍为"老王"，询问助手是谁；助手回应自己是"小王"。随后用户表示很高兴认识助手。

1
interrupt_on={
2
    "get_weather": True,
3
    "read_email_tool": False,
4
    "send_email_tool": {
5
        "allowed_decisions": ["approve", "reject"],
6
    },
7
}

1
"get_weather": {
2
    "allowed_decisions": ["approve", "edit", "reject"],
3
}

1
from langchain.chat_models import init_chat_model
2
from dotenv import load_dotenv
3
import os
4

5
# 从.env文件中加载环境变量
6
load_dotenv(override=True)
7

8
CLOSEAI_API_KEY = os.getenv("CLOSEAI_API_KEY")
9
CLOSEAI_BASE_URL = os.getenv("CLOSEAI_BASE_URL")
10

11
model = init_chat_model(
12
    model="gpt-5.4-mini",
13
    model_provider="openai",
14
    api_key=CLOSEAI_API_KEY,
15
    base_url=CLOSEAI_BASE_URL
16
)

1
from langchain.agents import create_agent
2
from langchain.agents.middleware import HumanInTheLoopMiddleware
3
from langgraph.checkpoint.memory import InMemorySaver
4
from langchain.messages import HumanMessage
5
from langchain.tools import tool
6
from langgraph.types import Command
7
from rich import print as rprint
8

9
@tool
10
def get_weather(city: str, is_forcast: bool = False) -> str:
11
    """
12
    查询指定城市天气
13
    Args:
14
        city: 城市名称
15
        is_forcast: 是否包含明日天气预报？
16
    """
17
    res = f"{city}今天天气不错"
18
    if is_forcast:
19
        res += "\n明天下雨"
20
    return res
21

22
@tool
23
def get_news() -> str:
24
    """
25
    查询当日新闻
26
    """
27
    return "中方三艘油轮通过霍尔木兹海峡"
28

29
@tool
30
def read_email_tool(email_id: str) -> str:
31
    """通过邮件ID读取内容的伪函数"""
32
    return f"邮件ID：{email_id}\n是空的"
33

34
@tool
35
def send_email_tool(recipient: str, subject: str, body: str) -> str:
36
    """发送邮件伪函数"""
37
    print(">>> 真的执行发送邮件工具了")
38
    return f"发送给 {recipient} 的邮件标题是：{subject}，内容：{body}"
39

40
agent = create_agent(
41
    model=model,
42
    tools=[get_weather, get_news, read_email_tool, send_email_tool],
43
    checkpointer=InMemorySaver(),
44
    middleware=[
45
        HumanInTheLoopMiddleware(
46
            interrupt_on={
47
                "get_weather": True,
48
                "get_news": True,
49
                "read_email_tool": False,
50
                "send_email_tool": {
51
                    "allowed_decisions": ["approve", "reject"],
52
                    "description": "发送邮件中断啦"
53
                },
54
            },
55
            description_prefix="中断啦"
56
        ),
57
    ],
58
)
59

60
config = {"configurable": {"thread_id": "1"}}
61

62
# 第一次调用：会暂停在发送邮件前
63
response = agent.invoke(
64
    {
65
        "messages": [
66
            HumanMessage(content="请帮我查询今天北京的天气"
67
                         "查询今日新闻"
68
                         "查看ID为 'sk2131421' 的邮件内容，"
69
                         "向15641685664@qq.com发送邮件，标题是'哈哈哈'，内容是：'你好啊'"
70
                         "同时做这四件事")
71
        ]
72
    },
73
    config=config,
74
)

1
print("==== 第一次 invoke 返回 ====")
2
print("========= 原始响应 =========")
3
rprint(response)
4
print("========= 美化输出 =========")
5
for msg in response["messages"]:
6
    msg.pretty_print()
7

8
# 关键：看中断信息
9
interrupts = response.get("__interrupt__", [])
10
print("========== interrupts ==========")
11
rprint(interrupts)
12

13
# print("==== 逐个打印 interrupt 请求 ====")
14
action_requests = interrupts[0].value["action_requests"]
15
for action_request in action_requests:
16
    rprint(action_request)

1
# 如果有中断，说明进入人在环了
2
weather_decision = {
3
    "type": "edit",
4
    "edited_action": {
5
        "name": "get_weather",
6
        "args": {"city": "中国北京市", "is_forcast": True}
7
    }
8
}
9
news_decision = {
10
    "type": "approve",
11
}
12
send_email_decision = {
13
    "type": "approve"
14
}
15
decisions = {
16
    "decisions": []
17
}
18
# 决策的顺序必须和返回的中断请求顺序一致
19
for action_request in action_requests:
20
    if action_request["name"] == "get_weather":
21
        decisions["decisions"].append(weather_decision)
22
    if action_request["name"] == "get_news":
23
        decisions["decisions"].append(news_decision)
24
    if action_request["name"] == "send_email_tool":
25
        decisions["decisions"].append(send_email_decision)
26

27
if interrupts:
28
    # 审批通过
29
    resumed_response = agent.invoke(
30
        Command(resume=decisions),
31
        config=config,  # 必须是同一个 thread_id
32
    )

1
print("==== 审批后继续执行 ====")
2
for msg in resumed_response["messages"]:
3
    msg.pretty_print()

1
>>> 真的执行发送邮件工具了
2
==== 审批后继续执行 ====
3
================================= Ai Message ==================================
4
Tool Calls:
5
  get_weather (call_vtxDHcWFeTZfdROfOwSFuXZY)
6
  Args:
7
    city: 中国北京市
8
    is_forcast: True
9
  get_news (call_t1AatnKbXgnez8JktmLGT9D3)
10
  read_email_tool (call_bVcNLUT9lI7UCwhyUGG5fCO5)
11
    email_id: sk2131421
12
  send_email_tool (call_AaHWhE0t93yn35swB7bZnYyo)
13
    recipient: 15641685664@qq.com
14
    subject: 哈哈哈
15
    body: 你好啊
16
================================= Tool Message =================================
17
Name: get_weather
18
中国北京市今天天气不错
19
明天下雨
20
================================= Tool Message =================================
21
Name: get_news
22
中方三艘油轮通过霍尔木兹海峡
23
================================= Tool Message =================================
24
Name: read_email_tool
25
邮件ID：sk2131421
26
是空的
27
================================= Tool Message =================================
28
Name: send_email_tool
29
发送给 15641685664@qq.com 的邮件标题是：哈哈哈，内容：你好啊

1
BUILTIN_DETECTORS: dict[str, Detector] = {
2
    "email": detect_email,
3
    "credit_card": detect_credit_card,
4
    "ip": detect_ip,
5
    "mac_address": detect_mac_address,
6
    "url": detect_url,
7
}

1
from langchain.chat_models import init_chat_model
2
from dotenv import load_dotenv
3
import os
4

5
# 从.env文件中加载环境变量
6
load_dotenv(override=True)
7

8
CLOSEAI_API_KEY = os.getenv("CLOSEAI_API_KEY")
9
CLOSEAI_BASE_URL = os.getenv("CLOSEAI_BASE_URL")
10

11
model = init_chat_model(
12
    model="gpt-5.4-mini",
13
    model_provider="openai",
14
    api_key=CLOSEAI_API_KEY,
15
    base_url=CLOSEAI_BASE_URL
16
)

1
from langchain.agents import create_agent
2
from langchain.agents.middleware import PIIMiddleware
3
from langchain.messages import HumanMessage
4

5
agent = create_agent(
6
    model=model,
7
    tools=[],
8
    middleware=[
9
        PIIMiddleware("email", strategy="redact", apply_to_input=True),
10
        PIIMiddleware("credit_card", strategy="mask", apply_to_input=True),
11
        PIIMiddleware("url", strategy="hash", apply_to_input=True),
12
        PIIMiddleware("mac_address", strategy="mask", apply_to_input=True),
13
        PIIMiddleware("ip", strategy="block", apply_to_input=True),
14
    ]
15
)
16

17
response = agent.invoke({
18
    "messages": [HumanMessage("""
19
        帮我向 156168188@qq.com 发送一封邮件
20
        同时查看银行卡号： 5105-1051-0510-5100 的余额
21
        访问 https://localhost:12345
22
        确认这是不是 MAC地址： 11-11-11-11-11-11
23
    """)]
24
})
25
for msg in response["messages"]:
26
    msg.pretty_print()
27

28
try:
29
    response1 = agent.invoke({
30
        "messages": [HumanMessage("看看这个 IP 能不能 ping 通：192.168.10.1")]
31
    })
32
except Exception as e:
33
    print('=' * 30, '-> 抛异常 <-', '=' * 30)
34
    print(f"检测到IP，抛出异常：{e}")

1
================================= Human Message =================================
2
帮我向 [REDACTED_EMAIL] 发送一封邮件
3
同时查看银行卡号： ****-****-****-5100 的余额
4
访问 <url_hash:dd5fc2a9>
5
确认这是不是 MAC地址： **-**-**-**-**-11
6
============================== -> 抛异常 <- ==============================
7
检测到IP，抛出异常：Detected 1 instance(s) of ip in text content

1
import re
2

3
# 自定义检测函数
4
def detect_phone_number(content: str):
5
    return [
6
        {
7
            "text": m.group(0),   # 提取出具体匹配到的 11 位数字文本（例如 "13800138000"）
8
            "start": m.start(),   # 这段数字在原文本中的"起始索引位置"（从 0 开始算）
9
            "end": m.end()        # 这段数字在原文本中的"结束索引位置"
10
        } for m in re.finditer(r"[0-9]{11}", content)
11
    ]

1
text = "尚硅谷的电话是13812345678，康师傅的电话是13987654321。"
2
result = detect_phone_number(text)
3
print(result)

1
[{'text': '13812345678', 'start': 7, 'end': 18}, {'text': '13987654321', 'start': 26, 'end': 37}]

1
from langchain.agents import create_agent
2
from langchain.agents.middleware import PIIMiddleware
3
from langchain.messages import HumanMessage
4

5
agent = create_agent(
6
    model=model,
7
    tools=[],
8
    middleware=[
9
        PIIMiddleware("api_key", strategy="hash", apply_to_input=True,
10
                      detector=r"sk-[a-zA-Z0-9]+"),
11
        PIIMiddleware("phone_number", strategy="mask", apply_to_input=True,
12
                      detector=detect_phone_number)
13
    ]
14
)
15

16
response = agent.invoke({
17
    "messages": [HumanMessage("""
18
        这是不是有效的 API_KEY： sk-awef23AFEfaafaefa
19
        帮我给这个号码打电话： 12345612345
20
        访问 https://localhost:12345
21
    """)]
22
})
23
for msg in response["messages"]:
24
    msg.pretty_print()

1
================================= Human Message =================================
2
这是不是有效的 API_KEY： <api_key_hash:6c678cc0>
3
帮我给这个号码打电话： ****2345
4
访问 https://localhost:12345

1
你的任务是否需要拆解？
2
├── 否 (比如：问答、翻译、单次函数调用) ──> 绝不需要，浪费算力
3
└── 是 (比如：写一个包含多文件的工程)
4
    └── 步骤是否多变且需要应对失败？
5
        ├── 否 (步骤完全固定，如 A->B->C) ──> 传统的 LangGraph 线性节点即可
6
        └── 是 (AI 需要边做边调计划) ──> 引入 TodoListMiddleware

1
def add(a: int, b: int) -> int:
2
    """返回两个整数的和"""
3
    return a - b

1
from my_add import add
2

3
def test_add():
4
    """测试加法功能"""
5
    assert add(2, 3) == 5
6
    assert add(-1, 1) == 0
7
    assert add(0, 0) == 0
8
    assert add(10, -5) == 5

1
conda activate langchain
2
cd .\todo_workspace\
3
pytest -q

1
F
2
[100%]
3
===================================================================== FAILURES =====================================================================
4
_____________________________________________________________________ test_add _____________________________________________________________________
5
def test_add():
6
    """测试加法功能"""
7
>       assert add(2, 3) == 5
8
E       assert -1 == 5
9
E        +  where -1 = add(2, 3)
10
test_my_add.py:5: AssertionError
11
============================================================= short test summary info ==============================================================
12
FAILED test_my_add.py::test_add - assert -1 == 5
13
1 failed in 0.11s

1
from langchain.chat_models import init_chat_model
2
from dotenv import load_dotenv
3
import os
4

5
# 从.env文件中加载环境变量
6
load_dotenv(override=True)
7

8
CLOSEAI_API_KEY = os.getenv("CLOSEAI_API_KEY")
9
CLOSEAI_BASE_URL = os.getenv("CLOSEAI_BASE_URL")
10

11
model = init_chat_model(
12
    model="gpt-5.4-mini",
13
    model_provider="openai",
14
    api_key=CLOSEAI_API_KEY,
15
    base_url=CLOSEAI_BASE_URL
16
)

1
from langchain.tools import tool
2
from pathlib import Path
3
import subprocess
4

5
WORKSPACE = Path("../todo_workspace")
6

7
@tool
8
def list_files(path: str = ".") -> str:
9
    """
10
    列出工作区指定目录下的文件和子目录。path 只能是相对路径。
11
    Args:
12
        path: 工作区下的相对路径，一定指向目录，默认为.，表示工作区根路径，不能访问工作区外的目录
13
    """
14
    target = (WORKSPACE / path).resolve()
15
    workspace_root = WORKSPACE.resolve()
16
    if not str(target).startswith(str(workspace_root)):
17
        return "错误：只允许访问工作区内的目录。"
18
    if not target.exists():
19
        return f"错误：目录不存在: {path}"
20
    if not target.is_dir():
21
        return f"错误：不是目录: {path}"
22
    items = sorted(target.iterdir(), key=lambda p: (p.is_file(), p.name.lower()))
23
    if not items:
24
        return f"目录为空: {path}"
25
    lines = []
26
    for item in items:
27
        rel = item.relative_to(workspace_root)
28
        kind = "[DIR]" if item.is_dir() else "[FILE]"
29
        lines.append(f"{kind} {rel.as_posix()}")
30
    return "\n".join(lines)
31

32
@tool
33
def read_file(path: str) -> str:
34
    """
35
    读取工作区中的文本文件内容。path 只能是相对路径。
36
    Args:
37
        path: 工作区内的文件名
38
    """
39
    file_path = (WORKSPACE / path).resolve()
40
    if not str(file_path).startswith(str(WORKSPACE.resolve())):
41
        return "错误：只允许读取工作区内的文件。"
42
    if not file_path.exists():
43
        return f"错误：文件不存在: {path}"
44
    return file_path.read_text(encoding="utf-8")
45

46
@tool
47
def write_file(path: str, content: str) -> str:
48
    """
49
    写入工作区中的文本文件。path 只能是相对路径。
50
    Args:
51
        path: 工作区内的文件名
52
        content: 写入文件的内容
53
    """
54
    file_path = (WORKSPACE / path).resolve()
55
    if not str(file_path).startswith(str(WORKSPACE.resolve())):
56
        return "错误：只允许写入工作区内的文件。"
57
    file_path.write_text(content, encoding="utf-8")
58
    return f"已写入文件: {path}"
59

60
@tool
61
def run_tests() -> str:
62
    """
63
    在工作区运行 pytest -q，并返回输出。
64
    不接收任何参数，返回格式为
65
    returncode=0|1
66
    STDOUT:
67
    STDERR:
68
    """
69
    try:
70
        result = subprocess.run(
71
            ["pytest", "-q"],
72
            cwd=str(WORKSPACE),
73
            capture_output=True,
74
            text=True,
75
            timeout=20,
76
        )
77
        return (
78
            f"returncode={result.returncode}\n\n"
79
            f"STDOUT:\n{result.stdout}\n\n"
80
            f"STDERR:\n{result.stderr}"
81
        )
82
    except Exception as e:
83
        return f"运行测试失败: {e}"

1
from langchain.agents import create_agent
2
from langchain.agents.middleware import TodoListMiddleware
3
from langchain.messages import HumanMessage
4
from rich import print as rprint
5

6
# 1. 初始化 Agent
7
agent = create_agent(
8
    model=model,
9
    # write_todos 等工具，TodoListMiddleware 需要配合这些工具使用
10
    tools=[list_files, read_file, write_file, run_tests],
11
    # 引入 Todo 列表中间件
12
    middleware=[TodoListMiddleware()],
13
    system_prompt=(
14
        "你是一个代码修复助手。遇到多步骤任务时，先使用 write_todos 制定待办事项；"
15
        "然后读取文件、修复代码并运行测试。工作全部在工作区下进行。"
16
    ),
17
)
18

19
# 2. 使用invoke进行同步调用
20
print("正在执行 Agent 任务...")
21
final_state = agent.invoke(
22
    {
23
        "messages": [
24
            HumanMessage(content="请测试并修复工作区下 my_add.py 文件中的代码")
25
        ]
26
    }
27
)
28

29
# 3. 直观展示中间件产生的数据结果
30
rprint(final_state)

1
正在执行 Agent 任务...
2
...（write_todos 规划四个步骤：检查工作区结构并定位 my_add.py → 阅读 my_add.py 及相关测试/调用代码确认问题 → 修复 my_add.py 中的代码缺陷 → 运行 pytest 验证修复结果）...
3
Updated todo list to [{'content': '检查工作区结构并定位 my_add.py', 'status': 'completed'}, {'content': '阅读 my_add.py 及相关测试/调用代码，确认问题', 'status': 'completed'}, {'content': '修复 my_add.py 中的代码缺陷', 'status': 'completed'}, {'content': '运行 pytest 验证修复结果', 'status': 'in_progress'}]

1
已修复 `my_add.py`，把减法改成了加法。
2

3
我尝试运行测试，但当前工作区环境里 `pytest` 命令不可用/找不到，因此无法在此环境中完成自动测试验证。
4

5
修复内容如下：
6

7
def add(a: int, b: int) -> int:
8
    """返回两个整数的和"""
9
    return a + b

1
'todos': [
2
    {'content': '检查工作区结构并定位 my_add.py', 'status': 'completed'},
3
    {'content': '阅读 my_add.py 及相关测试/调用代码，确认问题', 'status': 'completed'},
4
    {'content': '修复 my_add.py 中的代码缺陷', 'status': 'completed'},
5
    {'content': '运行 pytest 验证修复结果', 'status': 'in_progress'}
6
]

1
[用户请求] -> "修复 my_add.py"
2
    │
3
    ▼
4
[Agent 思考] -> 意识到是多步骤复杂任务
5
    │
6
    ▼
7
[触发工具] -> 调用 write_todos(tasks=[...])
8
    │
9
    ┌─┴────────────────────────┐
10
    │ TodoListMiddleware 拦截 │-> 自动解析工具参数，更新 State 中的 {"todos": [...]}
11
    └─┬────────────────────────┘
12
    │
13
    ▼
14
[继续执行] -> 读取文件、修改、测试...
15
    │
16
    ▼
17
[最终返回] -> final_state 携带了被中间件更新过的最新 "todos" 列表