LANGCHAIN MODULE 3 · LESSON 3.5 Email Agent

1 项目目标和运行流程

这个 notebook 是 Module 3 的综合项目。Agent 要先认证用户，认证成功后检查收件箱，再根据用户指令起草邮件，最后在发送前暂停等待人类审批。

1. Auth

用户提交邮箱和密码。

2. Tools unlock

认证成功后开放收件箱和发邮件工具。

3. Inbox

Agent 读取近期邮件。

4. HITL

发送邮件前等待审批。

2 Context 保存正确凭证，State 保存认证状态

正确邮箱和密码通过 runtime context 传入；是否已经认证则写进 Agent state。这样同一个 thread 后续调用可以知道用户已登录。

Python

@dataclass
class EmailContext:
    email_address: str = "julie@example.com"
    password: str = "password123"

class AuthenticatedState(AgentState):
    authenticated: bool

数据	放在哪里	原因
正确邮箱/密码	Runtime Context	由外部系统提供，不需要模型长期保存
authenticated	Agent State	认证结果要跨同一个 thread 后续步骤保留

3 定义认证、查收件箱、发邮件工具

authenticate 会比较用户输入和 runtime context。如果成功，返回 Command(update=...) 把 authenticated 写入 state。

Python

@tool
def check_inbox() -> str:
    """Check the inbox for recent emails"""
    return """
    Hi Julie,
    I'm going to be in town next week and was wondering if we could grab a coffee?
    - best, Jane (jane@example.com)
    """

@tool
def send_email(to: str, subject: str, body: str) -> str:
    """Send an response email"""
    return f"Email sent to {to} with subject {subject} and body {body}"

@tool
def authenticate(email: str, password: str, runtime: ToolRuntime) -> Command:
    """Authenticate the user with the given email and password"""
    if email == runtime.context.email_address and password == runtime.context.password:
        return Command(update={
            "authenticated": True,
            "messages": [ToolMessage("Successfully authenticated", tool_call_id=runtime.tool_call_id)],
        })
    else:
        return Command(update={
            "authenticated": False,
            "messages": [ToolMessage("Authentication failed", tool_call_id=runtime.tool_call_id)],
        })

认证结果写 state

认证不是只返回一句话，而是改变 Agent 后续可用工具和 prompt 的状态。

4 动态工具：认证前后开放不同工具

认证前只允许调用 authenticate。认证后开放 check_inbox 和 send_email。这是动态工具权限的实战版。

Python

@wrap_model_call
def dynamic_tool_call(
    request: ModelRequest,
    handler: Callable[[ModelRequest], ModelResponse],
) -> ModelResponse:
    """Allow read inbox and send email tools only if user provides correct email and password"""
    authenticated = request.state.get("authenticated")

    if authenticated:
        tools = [check_inbox, send_email]
    else:
        tools = [authenticate]

    request = request.override(tools=tools)
    return handler(request)

不要只靠 prompt

如果未认证用户仍能看到 send_email 工具，就存在误调用风险。这里从工具列表层面限制能力。

5 动态 Prompt：认证前后换角色说明

Prompt 也跟随认证状态变化：未认证时只做认证助手；认证后变成可以查收件箱和发送邮件的助手。

Python

from langchain.agents.middleware import dynamic_prompt

authenticated_prompt = """You are a helpful assistant that can check the inbox and send emails.
Your first step after authentication is to check the inbox."""
unauthenticated_prompt = "You are a helpful assistant that can authenticate users."

@dynamic_prompt
def dynamic_prompt(request: ModelRequest) -> str:
    """Generate system prompt based on authentication status"""
    authenticated = request.state.get("authenticated")

    if authenticated:
        return authenticated_prompt
    else:
        return unauthenticated_prompt

工具和 prompt 配合

工具控制“能做什么”，prompt 控制“应该怎么做”。认证状态变化时，两者都同步变化。

6 发送邮件前加入人工审批

即使用户已认证，发邮件仍然有外部副作用，所以 send_email 继续受 HITL 控制。

Python

humanInTheLoopMiddleware = HumanInTheLoopMiddleware(
    interrupt_on={
        "authenticate": False,
        "check_inbox": False,
        "send_email": True,
    },
    description_prefix="Tool execution requires approval",
)

agent = create_agent(
    deepseek_model(),
    tools=[authenticate, check_inbox, send_email],
    checkpointer=InMemorySaver(),
    state_schema=AuthenticatedState,
    context_schema=EmailContext,
    middleware=[dynamic_tool_call, dynamic_prompt, humanInTheLoopMiddleware],
)

多层防护

认证控制工具可见性，HITL 控制高风险工具执行。两者解决的是不同风险。

7 运行：登录、草拟、审批

第一次调用提交凭证；第二次让 Agent 草拟回复；当它准备发送邮件时，返回 __interrupt__。审批通过后，用同一个 thread 恢复执行。

Python

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

response = agent.invoke(
    {"messages": [HumanMessage(content="julie@example.com, password123")]},
    context=EmailContext(),
    config=config,
)

response = agent.invoke(
    {"messages": [HumanMessage(content="any draft is fine. don't check back.")]},
    context=EmailContext(),
    config=config,
)

print(response["__interrupt__"][0].value["action_requests"][0]["args"]["body"])

response = agent.invoke(
    Command(resume={"decisions": [{"type": "approve"}]}),
    config=config,
)

本课总结

这个 Agent 同时使用了 runtime context、state、dynamic tools、dynamic prompt、HITL 和 checkpointer，是 Module 3 的完整闭环。