LangChain 入门：构建第一个 Agent

本文目标

LangChain 1.x 之后，构建 Agent 的入口收敛成一个函数：create_agent。本文基于官方 LangChain Academy 的 Introduction to LangChain (Python) 课程和配套仓库 lca-lc-foundations，把三个模块串成一条学习路径，可以让你了解一些基本概念吧：

• Module 1 基础：模型与消息、工具、提示词与结构化输出、记忆
• Module 2 进阶：状态与运行时上下文、MCP、多 Agent 协作
• Module 3 控制：消息管理、动态模型 / 提示 / 工具、人工审批，最后合成一个邮件 Agent

课程默认用 OpenAI，本文全部换成 DeepSeek，并且是在本地 SOCKS 代理下实验的，所以多出两件事：安装 langchain-deepseek，以及给 httpx 装上 SOCKS 支持。DeepSeek 与课程示例的行为差异（主要是结构化输出）也一并记录。文中代码我都在 DeepSeek 上跑过，未跑过的会明确说明。

Agent 基本概念

四个核心对象

• 模型（Model）：负责理解和决策的大模型。LangChain 用统一接口封装不同厂商，换模型只改一个字符串。
• 消息（Message）：对话的基本单位，分 HumanMessage、AIMessage、ToolMessage 等。Agent 的全部对话状态本质上就是一个消息列表。
• 工具（Tool）：模型可以调用的函数。模型自己不执行，只输出“调用哪个工具、参数是什么”。
• Agent：把模型和工具装进一个循环：模型决策、按需调用工具、拿到结果后继续，直到给出最终答复。

执行循环

Agent 与单次模型调用的区别就在这个循环：

flowchart LR
    A[用户消息] --> B[模型决策]
    B -->|需要工具| C[执行工具]
    C -->|结果写回消息| B
    B -->|无需工具| D[最终答复]

每轮循环里，工具调用请求和工具结果都作为消息追加到列表，下一轮模型能看到完整过程。后文的状态、中间件、人工审批，本质上都是在这个循环的不同位置插入控制逻辑。

环境准备

依赖安装

课程要求 Python ≥ 3.12 且 < 3.14，推荐用 uv：

git clone --depth 1 https://github.com/langchain-ai/lca-lc-foundations.git
cd lca-lc-foundations
cp example.env .env

uv sync                      # 课程原有依赖
uv add langchain-deepseek    # DeepSeek 模型集成
uv add "httpx[socks]"        # SOCKS 代理支持（见下节）

课程依赖里只有 langchain-openai、langchain-anthropic 等，没有 DeepSeek 集成，需要手动加 langchain-deepseek。我实验时的版本：langchain 1.3.14、langchain-deepseek 1.1.1、langgraph 1.2.10、Python 3.13。

密钥配置

.env 里把 OPENAI_API_KEY 换成 DeepSeek 的：

DEEPSEEK_API_KEY='你的 DeepSeek Key'
TAVILY_API_KEY='你的 Tavily Key'   # 联网搜索工具用

代码统一用 python-dotenv 的 load_dotenv() 加载。.env 已在课程仓库的 .gitignore 里，不要提交。

课程笔记本里有不少单元格硬编码了 gpt-5-nano、claude-sonnet-4-6，需要手动换成 DeepSeek 模型名，否则会因缺少对应 Key 报错。

模型与消息

init_chat_model 按模型名前缀推断厂商，deepseek 开头的会使用 langchain-deepseek：

from dotenv import load_dotenv
from langchain.chat_models import init_chat_model

load_dotenv()

model = init_chat_model("deepseek-flash", temperature=1.0)  # 额外参数透传给模型
response = model.invoke("What's the capital of the Moon?")

print(response.content)            # 回答正文
print(response.response_metadata)  # token 用量、模型名、finish_reason

invoke 返回 AIMessage。DeepSeek 思考模式的推理过程放在 additional_kwargs["reasoning_content"]，不会混进 content；response_metadata["token_usage"] 含推理 token 数，排查成本时有用。

没有工具时，Agent 就是“带消息列表的模型调用”：

from langchain.agents import create_agent
from langchain.messages import HumanMessage

agent = create_agent(model="deepseek-flash")
response = agent.invoke(
    {"messages": [HumanMessage(content="What's the capital of the Moon?")]}
)
print(response["messages"][-1].content)  # 返回的 messages 含完整历史，答案在最后一条

输入输出都是 {"messages": [...]} 字典。长回答可以流式输出：

for token, metadata in agent.stream(
    {"messages": [HumanMessage(content="Tell me all about Luna City")]},
    stream_mode="messages",  # 以消息块为粒度流式返回
):
    if token.content:        # 过滤没有正文的块（如工具调用块）
        print(token.content, end="", flush=True)

工具调用

用 @tool 把普通函数变成工具。类型注解和 docstring 就是给模型看的说明书，写不清楚模型就会乱用：

from langchain.agents import create_agent
from langchain.messages import HumanMessage
from langchain.tools import tool

@tool("square_root", description="Calculate the square root of a number")
def square_root(x: float) -> float:
    return x ** 0.5

print(square_root.invoke({"x": 467}))  # 工具可以脱离 Agent 单独测试

agent = create_agent(
    model="deepseek-flash",
    tools=[square_root],   # 把 square_root 传入
    system_prompt="You are an arithmetic wizard. Use your tools to calculate square roots.",
)
response = agent.invoke(
    {"messages": [HumanMessage(content="What is the square root of 467?")]}
)

我的实际输出里，response["messages"] 是四条，正好对应循环图：

1. HumanMessage：用户问题
2. AIMessage：content 为空，tool_calls 为 {'name': 'square_root', 'args': {'x': 467}, 'id': 'call_...'}
3. ToolMessage：工具返回 21.61018278497431，通过 tool_call_id 与上一条配对
4. AIMessage：模型据此给出最终答复

调试时直接打印 response["messages"]，看模型调用了什么、传了什么参数，比反复改提示词有效。

提示词与结构化输出

系统提示词

system_prompt 决定角色与边界。课程演示了递进写法：纯角色描述、加 few-shot 示例、规定输出格式。示例比抽象描述更能约束风格，格式要求越具体越稳定。但文本格式始终是“尽量遵守”，程序要消费结果就该用结构化输出。

结构化输出

给 create_agent 传 Pydantic 模型，结果出现在 response["structured_response"]：

from langchain.agents import create_agent
from pydantic import BaseModel

class CapitalInfo(BaseModel):
    name: str
    location: str
    vibe: str
    economy: str

agent = create_agent(
    model="deepseek-flash",
    system_prompt="You are a science fiction writer, create a capital city at the users request.",
    response_format=CapitalInfo,   # 课程写法：直接传 schema
)

这是课程写法，在我的 DeepSeek 环境里直接报错：

BadRequestError: Error code: 400 - This response_format type is unavailable now

直接传 schema 时 LangChain 默认用厂商原生的结构化输出，我这边的 DeepSeek 接口不支持。改用 ToolStrategy（把 schema 包装成“工具”让模型调用）后换成另一个错误：Thinking mode does not support this tool_choice——它会强制调用该工具，而思考模式不允许。最终可行的是关闭思考模式再配 ToolStrategy：

from langchain.agents import create_agent
from langchain.agents.structured_output import ToolStrategy
from langchain.chat_models import init_chat_model
from langchain.messages import HumanMessage
from pydantic import BaseModel

class CapitalInfo(BaseModel):
    name: str
    location: str
    vibe: str
    economy: str

model = init_chat_model(
    "deepseek-flash",
    extra_body={"thinking": {"type": "disabled"}},  # 关闭思考模式，透传给接口
)

agent = create_agent(
    model=model,
    system_prompt="You are a science fiction writer, create a capital city at the users request.",
    response_format=ToolStrategy(CapitalInfo),
)
info = agent.invoke(
    {"messages": [HumanMessage(content="What is the capital of The Moon?")]}
)["structured_response"]
print(info.name, info.location)

这套组合实测通过。接口表现可能随服务商和版本变化，如果你的环境没有这两个限制，直接传 schema 即可。

记忆

Agent 默认无状态，每次 invoke 都是新对话。加记忆只需两步：配置 checkpointer，调用时指定 thread_id：

from langchain.agents import create_agent
from langchain.messages import HumanMessage
from langgraph.checkpoint.memory import InMemorySaver

agent = create_agent("deepseek-flash", checkpointer=InMemorySaver())
config = {"configurable": {"thread_id": "1"}}  # 同一个 thread_id 共享历史

agent.invoke(
    {"messages": [HumanMessage(content="Hello my name is Seán and my favourite colour is green")]},
    config,
)
response = agent.invoke(
    {"messages": [HumanMessage(content="What's my favourite colour?")]},
    config,
)
print(response["messages"][-1].content)  # 回答 green

实测第二轮返回的 messages 共 4 条，历史被自动带上了。要点：

• thread_id 区分会话：换 id 就是全新对话，多用户场景用用户或会话 id。
• InMemorySaver 只存在进程内存，重启即丢。生产环境换 LangGraph 的持久化 checkpointer，具体包名以官方文档为准。
• 历史随轮数增长，成本和上下文长度都会涨，需要裁剪或摘要，见中间件一章。

状态与上下文

消息之外，Agent 还需要保存业务数据，也需要接收“本次调用的固定信息”。LangChain 把这两者分开：状态会变，上下文不变。

自定义状态

继承 AgentState 加字段，工具通过返回 Command(update=...) 写入状态，通过 runtime.state 读取：

from langchain.agents import AgentState, create_agent
from langgraph.checkpoint.memory import InMemorySaver
from langchain.tools import tool, ToolRuntime
from langchain.messages import HumanMessage, ToolMessage
from langgraph.types import Command

class CustomState(AgentState):
    favourite_colour: str

@tool
def update_favourite_colour(favourite_colour: str, runtime: ToolRuntime) -> Command:
    """Update the favourite colour of the user in the state once they've revealed it."""
    return Command(update={
        "favourite_colour": favourite_colour,
        "messages": [ToolMessage(                    # 写状态时必须同时回一条 ToolMessage
            "Successfully updated favourite colour",
            tool_call_id=runtime.tool_call_id)],
    })

agent = create_agent(
    "deepseek-flash",
    tools=[update_favourite_colour],
    checkpointer=InMemorySaver(),
    state_schema=CustomState,
)
response = agent.invoke(
    {"messages": [HumanMessage(content="My favourite colour is green")]},
    {"configurable": {"thread_id": "1"}},
)
print(response["favourite_colour"])  # green，实测通过

状态字段也可以在调用时直接传入，作为初始值。

response = agent.invoke(
    {
        "messages": [HumanMessage(content="What's my favourite colour?")],
        "favourite_colour": "green",   # 初始值，直接写进输入字典
    },
    {"configurable": {"thread_id": "1"}},
)

运行时上下文

上下文用来传递用户身份、语言偏好等不由模型决定、调用期间不变的信息。定义一个 dataclass，通过 context_schema 声明，调用时传入，工具里用 ToolRuntime[类型] 读取：

from dataclasses import dataclass
from langchain.agents import create_agent
from langchain.messages import HumanMessage
from langchain.tools import tool, ToolRuntime

@dataclass
class ColourContext:
    favourite_colour: str = "blue"

@tool
def get_favourite_colour(runtime: ToolRuntime[ColourContext]) -> str:
    """Get the favourite colour of the user"""
    return runtime.context.favourite_colour

agent = create_agent(
    model="deepseek-flash",
    tools=[get_favourite_colour],
    context_schema=ColourContext,
)
response = agent.invoke(
    {"messages": [HumanMessage(content="What is my favourite colour?")]},
    context=ColourContext(favourite_colour="green"),  # 本次调用的上下文
)

实测回答为 green。两者的取舍：模型需要“记住并可能更新”的放状态，调用方在入口就确定、模型不该改的放上下文。

MCP 接入工具

MCP（Model Context Protocol）是把外部能力以统一协议暴露给 Agent 的标准。有了它，工具不必写在 Agent 代码里，可以是别人提供的服务。依赖是 langchain-mcp-adapters（课程已包含）。

用官方 FastMCP 写一个本地服务：

# server.py
from mcp.server.fastmcp import FastMCP

mcp = FastMCP("demo")

@mcp.tool()
def add(a: int, b: int) -> int:
    """Add two integers"""
    return a + b

if __name__ == "__main__":
    mcp.run(transport="stdio")  # 通过标准输入输出与客户端通信

Agent 侧用 MultiServerMCPClient 连接并取回工具，注意 MCP 工具是异步的，要用 ainvoke：

import asyncio
from langchain.agents import create_agent
from langchain.messages import HumanMessage
from langchain_mcp_adapters.client import MultiServerMCPClient

async def main():
    client = MultiServerMCPClient({
        "demo": {"transport": "stdio", "command": "python", "args": ["server.py"]},
    })
    tools = await client.get_tools()          # MCP 工具直接就是 LangChain 工具
    agent = create_agent("deepseek-flash", tools=tools)
    r = await agent.ainvoke({"messages": [HumanMessage(content="What is 123 + 456?")]})
    print(r["messages"][-1].content)

asyncio.run(main())

我实测通过，模型调用 add 并回答 579。要点：

• 传输方式：stdio 由客户端拉起本地子进程（课程里还用 uv run python -m mcp_server_time 启动现成的时间服务）；streamable_http 连远程地址，课程旅行 Agent 用的是 https://mcp.kiwi.com 查机票。
• 不只有工具：MCP 还能提供资源（get_resources）和提示词（get_prompt），课程把服务端的提示词直接作为 Agent 的 system_prompt。
• 容错：远程 MCP 服务可能超时或返回异常。课程 2.4 用 tool_interceptors 写了一个重试拦截器，对可重试错误做指数退避，其余错误作为文本返回给 Agent，让它自己调整重试，而不是让整个流程崩掉。这是个值得照搬的原则：工具错误尽量返回给 Agent，而不是抛出。
• 课程的 Jupyter 里可以直接 await；写成脚本要像上面这样套 asyncio.run。

多 Agent 协作

当任务涉及多个领域，把所有工具塞给一个 Agent 会让提示词臃肿、选择出错。课程的做法是子 Agent 作为工具：每个子 Agent 专注一个领域，主 Agent 只负责分派。

from langchain.agents import create_agent
from langchain.messages import HumanMessage
from langchain.tools import tool

@tool
def square_root(x: float) -> float:
    """Calculate the square root of a number"""
    return x ** 0.5

@tool
def square(x: float) -> float:
    """Calculate the square of a number"""
    return x ** 2

subagent_1 = create_agent(model="deepseek-flash", tools=[square_root])
subagent_2 = create_agent(model="deepseek-flash", tools=[square])

@tool
def call_subagent_1(x: float) -> str:
    """Call subagent 1 in order to calculate the square root of a number"""
    r = subagent_1.invoke({"messages": [HumanMessage(content=f"Calculate the square root of {x}")]})
    return r["messages"][-1].content   # 只把最终答复交还给主 Agent

@tool
def call_subagent_2(x: float) -> str:
    """Call subagent 2 in order to calculate the square of a number"""
    r = subagent_2.invoke({"messages": [HumanMessage(content=f"Calculate the square of {x}")]})
    return r["messages"][-1].content

main_agent = create_agent(
    model="deepseek-flash",
    tools=[call_subagent_1, call_subagent_2],
    system_prompt="You are a helpful assistant who can call subagents to calculate the square root or square of a number.",
)

我用同样结构实测，主 Agent 的消息序列是 Human → AI（调用子 Agent 工具）→ Tool → AI，与单个工具调用完全一致。好处有三个：

• 上下文隔离：子 Agent 内部的多轮工具调用不会污染主 Agent 的消息列表。
• 职责清晰：每个子 Agent 有独立的提示词和工具集，容易调试。
• 可组合：子 Agent 可以是任何 Agent，包括接了 MCP 的。

课程 2.4 的婚礼策划是完整案例：协调者 Agent 先用 update_state 把出发地、目的地、宾客数、音乐风格写入状态，再把航班（MCP 旅行服务）、场地（网页搜索）、歌单（SQL 查询）分派给三个子 Agent，工具通过 runtime.state 取参数。该案例需要 Kiwi MCP 与 Tavily，我没有完整跑通，只验证了其中的状态、子 Agent 与 MCP 各个环节。

中间件

中间件是在 Agent 循环的关键位置插入逻辑的机制，课程 Module 3 围绕它展开。模型调用前后、整个 Agent 开始前，都有对应的钩子。

消息摘要与裁剪

前面说过历史会无限增长。SummarizationMiddleware 在超过阈值时，把早期消息压缩成摘要：

from langchain.agents import create_agent
from langchain.agents.middleware import SummarizationMiddleware
from langgraph.checkpoint.memory import InMemorySaver

agent = create_agent(
    model="deepseek-flash",
    checkpointer=InMemorySaver(),
    middleware=[
        SummarizationMiddleware(
            model="deepseek-flash",     # 生成摘要的模型，可以用更便宜的，不过 deepseek-flash 本来也不贵
            trigger=("tokens", 100),    # 超过 100 token 触发（课程为演示设得很小）
            keep=("messages", 1),       # 摘要之外保留最近 1 条原始消息
        )
    ],
)

需要自定义逻辑时，用 @before_agent 等装饰器自己写。课程示例是在 Agent 运行前删除所有 ToolMessage：

from langchain.agents import AgentState
from langchain.agents.middleware import before_agent
from langchain.messages import RemoveMessage, ToolMessage
from langgraph.runtime import Runtime

@before_agent
def trim_messages(state: AgentState, runtime: Runtime):
    """Remove all the tool messages from the state"""
    tool_msgs = [m for m in state["messages"] if isinstance(m, ToolMessage)]
    return {"messages": [RemoveMessage(id=m.id) for m in tool_msgs]}

摘要中间件我只跑通了装配和调用，用很短的对话没能触发压缩，所以没有验证压缩后的效果。

动态提示词

@dynamic_prompt 根据状态或上下文在每次模型调用前生成系统提示。例如按用户语言切换：

from dataclasses import dataclass
from langchain.agents import create_agent
from langchain.agents.middleware import dynamic_prompt, ModelRequest
from langchain.messages import HumanMessage

@dataclass
class LanguageContext:
    user_language: str = "English"

@dynamic_prompt
def user_language_prompt(request: ModelRequest) -> str:
    lang = request.runtime.context.user_language
    base = "You are a helpful assistant."
    return base if lang == "English" else f"{base} only respond in {lang}."

agent = create_agent(
    model="deepseek-flash",
    context_schema=LanguageContext,
    middleware=[user_language_prompt],
)
agent.invoke(
    {"messages": [HumanMessage(content="Hello, how are you?")]},
    context=LanguageContext(user_language="French"),
)

动态模型与工具

@wrap_model_call 包裹整个模型调用，可以通过 request.override(...) 替换模型或工具。动态模型的课程例子：对话短用便宜模型，超过 10 条消息换大模型。动态工具的例子：按用户角色限制可用工具：

from dataclasses import dataclass
from typing import Callable
from langchain.agents import create_agent
from langchain.agents.middleware import wrap_model_call, ModelRequest, ModelResponse

@dataclass
class UserRole:
    user_role: str = "external"

@wrap_model_call
def dynamic_tool_call(request: ModelRequest,
                      handler: Callable[[ModelRequest], ModelResponse]) -> ModelResponse:
    if request.runtime.context.user_role != "internal":
        request = request.override(tools=[web_search])  # 外部用户只能用搜索
    return handler(request)

agent = create_agent(
    model="deepseek-flash",
    tools=[web_search, sql_query],
    middleware=[dynamic_tool_call],
    context_schema=UserRole,
)

我实测 external 角色提问“数据库里有多少艺术家”时，模型没有 SQL 工具可用，直接回答自己没有数据库访问权限。这是比提示词更可靠的权限控制：模型看不见的工具，就不可能被它调用。

人工审批

涉及发邮件、转账、删数据这类不可逆操作，应该让人确认后再执行。HumanInTheLoopMiddleware 在指定工具调用前暂停 Agent，等待决定，这需要 checkpointer 保存暂停点：

from langchain.agents import create_agent
from langchain.agents.middleware import HumanInTheLoopMiddleware
from langchain.messages import HumanMessage
from langgraph.checkpoint.memory import InMemorySaver
from langgraph.types import Command

agent = create_agent(
    model="deepseek-flash",
    tools=[read_email, send_email],
    checkpointer=InMemorySaver(),
    middleware=[HumanInTheLoopMiddleware(
        interrupt_on={"read_email": False, "send_email": True},  # 只拦截发送
    )],
)
config = {"configurable": {"thread_id": "1"}}

response = agent.invoke({"messages": [HumanMessage(content="...")]}, config=config)

# 暂停后，待审批的调用在 __interrupt__ 里
req = response["__interrupt__"][0].value["action_requests"][0]
print(req["name"], req["args"])

# 同一个 thread_id 恢复执行
agent.invoke(Command(resume={"decisions": [{"type": "approve"}]}), config=config)

decisions 有三种：

• approve：原样执行。
• reject：拒绝，可带 message 说明原因，Agent 会据此调整。
• edit：改写参数后执行，用 edited_action 给出工具名和新参数。

我实测了 approve：send_email 被拦截，参数出现在 __interrupt__，恢复后正常执行。reject 与 edit 来自课程示例，我没单独跑。

综合：邮件 Agent

课程最后用一个邮件 Agent 把前面的东西合起来：未登录时只能调用 authenticate，认证后才开放读信与发信，发信前要人工批准。它用到了自定义状态、运行时上下文、工具写状态、动态工具、动态提示词和人工审批：

# email_agent.py
from dataclasses import dataclass
from typing import Callable

from dotenv import load_dotenv
from langchain.agents import AgentState, create_agent
from langchain.agents.middleware import (
    HumanInTheLoopMiddleware, ModelRequest, ModelResponse,
    dynamic_prompt, wrap_model_call,
)
from langchain.messages import HumanMessage, ToolMessage
from langchain.tools import ToolRuntime, tool
from langgraph.checkpoint.memory import InMemorySaver
from langgraph.types import Command

load_dotenv()

@dataclass
class EmailContext:              # 运行时上下文：本次调用的固定凭据
    email_address: str = "[email protected]"
    password: str = "password123"

class AuthState(AgentState):     # 状态：会随对话变化
    authenticated: bool

@tool
def authenticate(email: str, password: str, runtime: ToolRuntime) -> Command:
    """Authenticate the user with the given email and password"""
    ok = (email == runtime.context.email_address
          and password == runtime.context.password)
    return Command(update={      # 工具写状态，同时必须回一条 ToolMessage
        "authenticated": ok,
        "messages": [ToolMessage(
            "Successfully authenticated" if ok else "Authentication failed",
            tool_call_id=runtime.tool_call_id)],
    })

@tool
def check_inbox() -> str:
    """Check the inbox for recent emails"""
    return "Hi Julie, can we grab a coffee next week? - Jane ([email protected])"

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

@wrap_model_call                 # 未认证只暴露 authenticate
def dynamic_tools(request: ModelRequest,
                  handler: Callable[[ModelRequest], ModelResponse]) -> ModelResponse:
    if request.state.get("authenticated"):
        tools = [check_inbox, send_email]
    else:
        tools = [authenticate]
    return handler(request.override(tools=tools))

@dynamic_prompt                  # 提示词也随认证状态切换
def dynamic_system_prompt(request: ModelRequest) -> str:
    if request.state.get("authenticated"):
        return ("You are a helpful assistant that can check the inbox and send emails. "
                "Your first step after authentication is to check the inbox.")
    return "You are a helpful assistant that can authenticate users."

agent = create_agent(
    "deepseek-flash",
    tools=[authenticate, check_inbox, send_email],
    checkpointer=InMemorySaver(),
    state_schema=AuthState,
    context_schema=EmailContext,
    middleware=[
        dynamic_tools,
        dynamic_system_prompt,
        HumanInTheLoopMiddleware(interrupt_on={"send_email": True}),  # 发信前必须人工批准
    ],
)

if __name__ == "__main__":
    cfg = {"configurable": {"thread_id": "1"}}
    ctx = EmailContext()
    r = agent.invoke({"messages": [HumanMessage("[email protected], password123")]}, context=ctx, config=cfg)
    print(r["messages"][-1].content)
    r = agent.invoke({"messages": [HumanMessage("Reply to Jane, any draft is fine. Don't check back.")]}, context=ctx, config=cfg)
    req = r["__interrupt__"][0].value["action_requests"][0]
    print("待审批:", req["name"], req["args"])
    r = agent.invoke(Command(resume={"decisions": [{"type": "approve"}]}), context=ctx, config=cfg)
    print(r["messages"][-1].content)

运行 uv run python email_agent.py，我实测的流程是：认证成功，Agent 读信后起草回信；send_email 被拦截并打印出收件人与正文；批准后邮件发出。模型的措辞每次不同，课程也注明过这一点，必要时调整第二条用户消息。

各模块的概念在这里的对应关系：

flowchart LR
    A[用户消息] --> B[动态提示词]
    B --> C[动态工具]
    C --> D[模型决策]
    D -->|send_email| E[人工审批]
    D -->|authenticate| F[写入状态]
    E -->|approve| G[执行工具]
    F --> B

认证工具写入状态后，下一轮循环的提示词和工具集自动切换，这就是“状态驱动行为”的完整例子。

小结与延伸

Agent 的构成一句话概括：模型 + 工具 + 消息列表，在循环里跑到不再需要工具为止。三个模块的递进关系是：

• Module 1 让 Agent 跑起来：模型、工具、提示词、结构化输出、记忆。
• Module 2 让 Agent 接入外部世界和分工：状态与上下文、MCP、多 Agent。
• Module 3 让 Agent 可控：消息管理、动态行为、人工审批。

课程还有几个本文没展开的内容：多模态消息（图片、音频）、基于 PDF 的 RAG 检索 Agent、用 langgraph dev 配合 Agent Chat UI 做本地调试、LangSmith 链路追踪。可以按需补充。

参考资料：

• LangChain Academy：Introduction to LangChain (Python) 课程
• 课程仓库：langchain-ai/lca-lc-foundations
• LangChain 官方文档：docs.langchain.com