LangChain 03：1.x Agent 教程

发表于 更新于 分类于
基于 LangChain 1.x 的 create_agent，覆盖工具、记忆、结构化输出、流式与中间件，示例对接火山方舟 Coding Plan。

前言

只做单次问答时，直接调用 Chat 模型往往就够用。
一旦需要查天气、算数、读写业务接口，模型必须在「推理 → 调工具 → 观察结果」之间循环，这就是 Agent。
LangChain 1.x 用 create_agent 把该循环固化在 LangGraph 运行时上，并统一挂载工具、系统提示、记忆与中间件。
本文以 LangChain 1.x 为准，围绕 Agent 实战展开；环境安装与 init_chat_model 入门见同系列《LangChain 入门教程》。
示例通过 OpenAI 兼容协议对接 火山方舟 Coding Plan：Base URL 为 https://ark.cn-beijing.volces.com/api/coding/v3，模型为 ark-code-latest。
请勿使用普通方舟 .../api/v3，以免无法抵扣 Coding Plan 额度。
下文每个 Python 示例都是完整可运行脚本：复制到项目根目录（与 .env 同级）后执行 uv run python xxx.py 即可。
下文需要 Python 3.12+，依赖用 uv 管理。

依赖

在项目目录声明 1.x 主包与 OpenAI 兼容集成（版本请按项目实际调整）。

1
uv add "langchain>=1.0,<2.0" langchain-openai langgraph python-dotenv rich

rich 用于在终端里更清楚地查看结构化结果：彩色高亮、自动缩进嵌套 dict / list / 消息对象，比内置 print 更适合调试 Agent 的 messages 与整次 invoke 返回值。
示例里常用 from rich import print as rprint，把原来的 print(...) 换成 rprint(...) 即可。

在项目根目录创建 .env，写入 Coding Plan 的 API Key 与专用 Base URL。
不要把 .env 提交进 Git。

1
2
OPENAI_API_KEY=你的火山方舟 API Key
OPENAI_BASE_URL=https://ark.cn-beijing.volces.com/api/coding/v3

示例脚本开头用 load_dotenv() 加载该文件。
langchain-openai 仍读取 OPENAI_* 环境变量；值必须来自 Coding Plan。
下文模型统一写 openai:ark-code-latest，也可在控制台改成具体模型名后同步替换代码。

运行示例可用 uv run python agent_demo.py，一般不必手动激活虚拟环境。

实现

工作原理

Agent 可理解为 Model + Harness：模型负责决策，Harness 负责在正确时机塞入提示、工具与状态。

一次典型循环如下。

  1. 模型阅读 messages（含系统提示）并决定是否调用工具。
  2. 若 AIMessage 含 tool_calls，运行时执行对应工具，把结果写成 ToolMessage。
  3. 模型再次阅读更新后的消息列表，直到不再发起工具调用，返回最终回复。

create_agent 返回的是已编译的 LangGraph 图，因此可直接 invoke / stream，也可用 checkpointer 做线程级记忆。

最小 Agent

最简用法只需模型标识、工具列表与可选系统提示。
普通函数只要写好类型注解与 docstring，即可作为工具传入；也可用 @tool 显式声明。

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
import os

from dotenv import load_dotenv
from langchain.agents import create_agent
from langchain.chat_models import init_chat_model
from langchain.tools import tool
from rich import print as rprint

load_dotenv()

api_key = os.environ["OPENAI_API_KEY"]
base_url = os.environ["OPENAI_BASE_URL"]

@tool
def get_weather(city: str) -> str:
    """查询指定城市的天气（示例数据）。"""
    return f"{city} 今天晴，25°C。"

model = init_chat_model(
    "openai:ark-code-latest",
    temperature=0,
    api_key=api_key,
    base_url=base_url,
)

agent = create_agent(
    model=model,
    tools=[get_weather],
    system_prompt="你是中文助手。需要天气信息时必须调用 get_weather，不要编造。",
)

result = agent.invoke({
    "messages": [{"role": "user", "content": "郑州今天天气怎么样？"}]
})
rprint(result["messages"][-1])

输入输出都以 messages 为中心；最后一条通常是最终 AIMessage。
若已导出 Coding Plan 的 OPENAI_BASE_URL，也可写 create_agent(model="openai:ark-code-latest", ...)。

定义工具

工具 docstring 会进入模型可见的 schema，写清楚「做什么、参数含义」比堆复杂实现更重要。
下面用加法与乘法两个工具跑通一次完整调用。

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
import os

from dotenv import load_dotenv
from langchain.agents import create_agent
from langchain.chat_models import init_chat_model
from langchain.tools import tool
from rich import print as rprint

load_dotenv()

api_key = os.environ["OPENAI_API_KEY"]
base_url = os.environ["OPENAI_BASE_URL"]

@tool
def add(a: float, b: float) -> float:
    """计算两个数的和。"""
    return a + b

@tool
def multiply(a: float, b: float) -> float:
    """计算两个数的乘积。"""
    return a * b

model = init_chat_model(
    "openai:ark-code-latest",
    temperature=0,
    api_key=api_key,
    base_url=base_url,
)

agent = create_agent(
    model=model,
    tools=[add, multiply],
    system_prompt="你是中文助手。算术必须调用 add 或 multiply，不要心算。",
)

result = agent.invoke({
    "messages": [{"role": "user", "content": "先算 3 加 5，再把结果乘以 2。"}]
})
rprint(result["messages"][-1])

不需要装饰器时，也可直接把带 docstring 的函数放进 tools=[...]，框架会自动包装。
敏感副作用（发邮件、删数据）不要只靠提示词约束，后面用中间件做人工审批更稳妥。

多工具协作

一次用户请求可能触发多次工具调用；Agent 会按需串联或并行选择工具，直到能给出答案。
下面把天气与计算器交给同一个 Agent。

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
import os

from dotenv import load_dotenv
from langchain.agents import create_agent
from langchain.chat_models import init_chat_model
from langchain.tools import tool
from rich import print as rprint

load_dotenv()

api_key = os.environ["OPENAI_API_KEY"]
base_url = os.environ["OPENAI_BASE_URL"]

@tool
def get_weather(city: str) -> str:
    """查询指定城市的天气（示例数据）。"""
    return f"{city} 今天晴，25°C。"

@tool
def add(a: float, b: float) -> float:
    """计算两个数的和。"""
    return a + b

model = init_chat_model(
    "openai:ark-code-latest",
    temperature=0,
    api_key=api_key,
    base_url=base_url,
)

agent = create_agent(
    model=model,
    tools=[get_weather, add],
    system_prompt="你是中文助手。天气用 get_weather，算术用 add，不要心算。",
)

result = agent.invoke({
    "messages": [{
        "role": "user",
        "content": "上海气温按示例是多少？再帮我算 19 加 6。",
    }]
})
rprint(result["messages"])

调试时可打印完整 result["messages"]，观察 AIMessage.tool_calls 与对应 ToolMessage 是否符合预期。

对话记忆

默认每次 invoke 互不共享历史。
要做多轮对话，给 Agent 配置 checkpointer，并在每次调用传入同一 thread_id。

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
import os

from dotenv import load_dotenv
from langchain.agents import create_agent
from langchain.chat_models import init_chat_model
from langgraph.checkpoint.memory import InMemorySaver
from rich import print as rprint

load_dotenv()

api_key = os.environ["OPENAI_API_KEY"]
base_url = os.environ["OPENAI_BASE_URL"]

model = init_chat_model(
    "openai:ark-code-latest",
    api_key=api_key,
    base_url=base_url,
)

agent = create_agent(
    model=model,
    tools=[],
    system_prompt="你是友好的中文助手，记得用户说过的话。",
    checkpointer=InMemorySaver(),
)

config = {"configurable": {"thread_id": "demo-user-1"}}

r1 = agent.invoke(
    {"messages": [{"role": "user", "content": "我叫小明，喜欢骑行。"}]},
    config,
)
rprint(r1["messages"][-1])

r2 = agent.invoke(
    {"messages": [{"role": "user", "content": "我叫什么？有什么爱好？"}]},
    config,
)
rprint(r2["messages"][-1])

thread_id 标识一条会话线程：同 ID 续聊，换 ID 即新会话。
InMemorySaver 仅适合本地演示；上线请换 SQLite / Postgres 等持久化 checkpointer。

结构化输出

需要把最终答案落成固定字段时，用 Pydantic 模型传给 response_format。
校验失败时框架会按策略重试，成功结果在 structured_response。

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
import os

from dotenv import load_dotenv
from langchain.agents import create_agent
from langchain.chat_models import init_chat_model
from langchain.tools import tool
from pydantic import BaseModel, Field
from rich import print as rprint

load_dotenv()

api_key = os.environ["OPENAI_API_KEY"]
base_url = os.environ["OPENAI_BASE_URL"]

class WeatherBrief(BaseModel):
    city: str = Field(description="城市名")
    summary: str = Field(description="一句话天气摘要")
    celsius: int = Field(description="气温（摄氏度）")

@tool
def get_weather(city: str) -> str:
    """查询指定城市的天气（示例数据）。"""
    return f"{city} 今天晴，25°C。"

model = init_chat_model(
    "openai:ark-code-latest",
    temperature=0,
    api_key=api_key,
    base_url=base_url,
)

agent = create_agent(
    model=model,
    tools=[get_weather],
    response_format=WeatherBrief,
    system_prompt="根据工具结果填写结构化天气摘要。",
)

result = agent.invoke({
    "messages": [{"role": "user", "content": "给我上海的天气摘要"}]
})
brief: WeatherBrief = result["structured_response"]
rprint(brief.model_dump())

下游 API、表单或数据库写入应优先读 structured_response，而不是再解析自然语言。

流式输出

长任务希望边跑边展示进度时，用 stream。
stream_mode="updates" 在每个 Agent 步骤后推送状态增量，便于展示「正在调工具 / 已得到结果」。

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
import os

from dotenv import load_dotenv
from langchain.agents import create_agent
from langchain.chat_models import init_chat_model
from langchain.tools import tool
from rich import print as rprint

load_dotenv()

api_key = os.environ["OPENAI_API_KEY"]
base_url = os.environ["OPENAI_BASE_URL"]

@tool
def get_weather(city: str) -> str:
    """查询指定城市的天气（示例数据）。"""
    return f"{city} 今天晴，25°C。"

model = init_chat_model(
    "openai:ark-code-latest",
    api_key=api_key,
    base_url=base_url,
)
agent = create_agent(
    model=model,
    tools=[get_weather],
    system_prompt="你是中文助手。需要天气时调用 get_weather。",
)

for chunk in agent.stream(
    {"messages": [{"role": "user", "content": "上海天气怎么样？"}]},
    stream_mode="updates",
):
    rprint(chunk)

若只要模型逐 token 输出，改用 stream_mode="messages"。

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
import os

from dotenv import load_dotenv
from langchain.agents import create_agent
from langchain.chat_models import init_chat_model
from langchain.tools import tool

load_dotenv()

api_key = os.environ["OPENAI_API_KEY"]
base_url = os.environ["OPENAI_BASE_URL"]

@tool
def get_weather(city: str) -> str:
    """查询指定城市的天气（示例数据）。"""
    return f"{city} 今天晴，25°C。"

model = init_chat_model(
    "openai:ark-code-latest",
    api_key=api_key,
    base_url=base_url,
)
agent = create_agent(
    model=model,
    tools=[get_weather],
    system_prompt="你是中文助手。",
)

for chunk in agent.stream(
    {"messages": [{"role": "user", "content": "用两句话介绍上海。"}]},
    stream_mode="messages",
):
    # 不同小版本 chunk 结构可能略有差异，按项目 langchain 版本对照官方 Streaming 文档
    token, metadata = chunk
    text = getattr(token, "content", None) or ""
    if text:
        print(text, end="", flush=True)
print()

较新的 1.x 小版本还提供 stream_events（事件投影 API）；若你安装的版本支持，可按官方 Streaming 文档改用 typed 投影消费。

中间件

中间件挂在 Agent 循环的钩子上，用于摘要、限流、人工审批、PII 脱敏等，而不必手写整张图。
对话变长时，用 SummarizationMiddleware 在触发条件满足后压缩历史，避免撑爆上下文窗口。

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
import os

from dotenv import load_dotenv
from langchain.agents import create_agent
from langchain.agents.middleware import SummarizationMiddleware
from langchain.chat_models import init_chat_model
from langgraph.checkpoint.memory import InMemorySaver
from rich import print as rprint

load_dotenv()

api_key = os.environ["OPENAI_API_KEY"]
base_url = os.environ["OPENAI_BASE_URL"]

model = init_chat_model(
    "openai:ark-code-latest",
    api_key=api_key,
    base_url=base_url,
)

agent = create_agent(
    model=model,
    tools=[],
    checkpointer=InMemorySaver(),
    middleware=[
        SummarizationMiddleware(
            model=model,
            trigger=("tokens", 4000),
            keep=("messages", 20),
        ),
    ],
    system_prompt="你是中文助手。",
)

config = {"configurable": {"thread_id": "agent-mw-summary-1"}}
result = agent.invoke(
    {"messages": [{"role": "user", "content": "用一句话介绍 LangChain Agent。"}]},
    config,
)
rprint(result["messages"][-1])

这里把同一个 Coding Plan 模型实例传给摘要中间件；也可再 init_chat_model 一个更轻量的模型做摘要。
trigger 控制何时摘要，keep 控制摘要后保留多少近期消息；阈值与条数可按模型窗口再调。

高风险工具可挂 HumanInTheLoopMiddleware：在真正执行前中断，等人工 approve / edit / reject。
下面示例先触发中断，再用 Command(resume=...) 自动批准，便于本地一次性跑通。

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
import os
import uuid

from dotenv import load_dotenv
from langchain.agents import create_agent
from langchain.agents.middleware import HumanInTheLoopMiddleware
from langchain.chat_models import init_chat_model
from langchain.tools import tool
from langgraph.checkpoint.memory import InMemorySaver
from langgraph.types import Command
from rich import print as rprint

load_dotenv()

api_key = os.environ["OPENAI_API_KEY"]
base_url = os.environ["OPENAI_BASE_URL"]

model = init_chat_model(
    "openai:ark-code-latest",
    temperature=0,
    api_key=api_key,
    base_url=base_url,
)

@tool
def send_email(to: str, subject: str, body: str) -> str:
    """发送邮件（示例：真正环境请接邮件服务）。"""
    return f"已发送给 {to}：{subject}"

@tool
def read_email(email_id: str) -> str:
    """读取邮件内容（示例数据）。"""
    return f"邮件 {email_id} 的正文……"

agent = create_agent(
    model=model,
    tools=[read_email, send_email],
    checkpointer=InMemorySaver(),
    middleware=[
        HumanInTheLoopMiddleware(
            interrupt_on={
                "send_email": {
                    "allowed_decisions": ["approve", "edit", "reject"],
                },
                "read_email": False,
            }
        ),
    ],
    system_prompt="你是邮件助手。用户要求发信时必须调用 send_email。",
)

config = {"configurable": {"thread_id": str(uuid.uuid4())}}

interrupted = agent.invoke(
    {
        "messages": [{
            "role": "user",
            "content": "给 demo@example.com 发一封主题为测试、正文为你好的邮件。",
        }]
    },
    config,
)
rprint(interrupted)

resumed = agent.invoke(
    Command(resume={"decisions": [{"type": "approve"}]}),
    config,
)
rprint(resumed["messages"][-1])

该能力依赖 checkpointer 保存中断状态；更完整的 UI 对接见同系列《LangChain 中间件教程》与官方 Human-in-the-loop 文档。

验证

按下面顺序确认 Agent 与 Coding Plan 路径可用。

  1. 执行 uv run python -c "import importlib.metadata as m; print(m.version('langchain'))"，主版本应为 1.x。
  2. 确认 .env 中 OPENAI_BASE_URL 为 https://ark.cn-beijing.volces.com/api/coding/v3。
  3. 逐段复制完整脚本到 .py 文件，用 uv run python 直接运行。
  4. 跑「最小 Agent」脚本，提问城市天气，回复应来自工具返回的示例文案。
  5. 跑「对话记忆」脚本，第二次提问应能答出「小明」与骑行相关爱好。
  6. 跑「结构化输出」脚本，打印的 dict 应含 city / summary / celsius 字段。
  7. 跑 stream_mode="updates"，终端应先后出现模型节点与工具节点相关增量。

若认证失败，检查 OPENAI_API_KEY 与 OPENAI_BASE_URL。
若模型名无效，确认套餐已开通 ark-code-latest，或改成控制台具体模型名。
若 HITL 的 Command(resume=...) 报错，对照当前小版本文档核对 resume 字段结构。

扩展

Agent 跑通后，可按场景继续加深。

  • 检索 Tool 化：把向量检索包成 @tool，由 Agent 决定何时查知识库。
  • 动态工具 / 动态模型：用中间件按用户角色裁剪可见工具，或按任务切换模型。
  • 持久化 checkpointer：用 SQLite / Postgres 替代 InMemorySaver，并按用户会话生成 thread_id。
  • 可观测性：接入 LangSmith 查看每轮工具参数与中间消息，便于调提示词。
  • Deep Agents：若需要规划、文件系统工具、子 Agent 等开箱组合，可在 create_agent 之上选用更高层封装。

总结

  1. 用 create_agent 组装 Coding Plan 模型、工具与系统提示，由 LangGraph 循环执行直到无更多 tool_calls。
  2. 工具靠清晰 docstring 与类型注解被模型正确选用；多工具场景打印完整 messages 便于排错。
  3. 多轮对话配置 checkpointer + 稳定的 thread_id；生产环境不要只用内存 saver。
  4. 需要固定字段时用 response_format 拿 structured_response；需要进度反馈时用 stream。
  5. 摘要、人工审批等横切能力优先挂中间件，而不是复制一套自定义 Agent 图。