LangChain Agents 开发：从 create_agent 到部署

# LangChain Agents 开发：从 create_agent 到部署

本篇目标

沿着真实开发顺序跑通一个 LangChain Agent，再逐步加入结构化输出、RAG、Memory、LangGraph、LangSmith 和部署。看完后不仅能分清生态组件，还能说明代码放在哪里、一次请求怎样运行以及复杂需求为什么要下沉到 LangGraph。

跑通完整 Agent接入 Tool 与 RAG掌握状态与编排形成面试表达

# 先记住一句话

LangChain 用来快速组装模型、Prompt、Tool 和标准 Agent 循环；LangGraph 用来显式编排复杂流程并保存运行状态；LangSmith 用来追踪、评测和部署 Agent。

LangChain：把常用零件组装成 Agent
     ↓ 复杂流程需要自己控制时
LangGraph：用状态、节点和连线编排执行过程
     ↓ 运行后产生轨迹和评测数据
LangSmith：观察问题、比较版本并部署服务

1
2
3
4
5

三者可以一起使用，也可以按需选择。一个简单工具型 Agent 只用 LangChain 就能运行；LangSmith 不是运行 Agent 的必需组件。

# 先看完整开发路径

这篇不从抽象名词开始，而是沿着下面这条路径展开：

明确任务
  └─ 创建 Python 项目
      └─ 选择并配置模型
          └─ 编写 System Prompt
              └─ 用 @tool 定义工具
                  └─ create_agent 组装并运行
                      ├─ 需要私有知识 → 加入 RAG
                      ├─ 需要多轮对话 → 加入 Checkpointer
                      ├─ 需要复杂编排 → 下沉 LangGraph
                      ├─ 需要质量闭环 → 接入 LangSmith
                      └─ 需要对外服务 → 部署 Agent Server 或自建 API

1
2
3
4
5
6
7
8
9
10
11

贯穿全文的例子是一个技术笔记问答助手：回答前可以搜索笔记，最后给出答案和来源；复杂版本还支持人工审核与断点恢复。

# 1. 项目启动

# 安装与目录

LangChain 核心包和模型提供方的集成包是分开的。下面使用 OpenAI 集成演示，Python 需要 3.10 及以上。

# 创建并进入项目目录。
mkdir interview-note-agent
cd interview-note-agent

# 创建虚拟环境，避免依赖污染系统 Python。
python -m venv .venv
source .venv/bin/activate

# 安装 Agent、OpenAI 集成、LangGraph、文本切分和环境变量依赖。
pip install -U langchain langchain-openai langgraph langchain-text-splitters python-dotenv

1
2
3
4
5
6
7
8
9
10

先建立一份很小但可以直接运行的目录：

interview-note-agent/
├─ .env.example          # 环境变量示例，不保存真实密钥
├─ app/
│  ├─ __init__.py        # 让 app 成为 Python 包
│  ├─ agent.py           # 组装并导出 Agent
│  └─ tools.py           # 定义 Agent 可以调用的工具
└─ run.py                # 本地运行入口

1
2
3
4
5
6
7

# 文件位置：.env.example
# 复制为 .env 后填入真实值，但不要把 .env 提交到 Git。
OPENAI_API_KEY=replace_with_your_key

# 模型名称由配置决定，代码不写死具体版本。
OPENAI_MODEL=replace_with_a_tool_calling_model

1
2
3
4
5
6

为什么模型名称不直接写死

模型可用性、价格和能力会变化。项目把模型名称放在环境变量中，既方便不同环境切换，也方便评测新旧模型；但每次切换仍要重新跑同一套评测，不能只看接口是否调用成功。

# 2. 从零写一个可运行的 Agent

# 先定义 Tool

Tool 是宿主应用提供给模型的能力。模型只负责提出调用请求，真正执行函数的是 Agent Runtime。

# 文件位置：app/tools.py
from langchain.tools import tool

# 这里只用内存数据演示完整链路；真实项目会替换为数据库或向量检索。
NOTES = [
    {
        "title": "Checkpoint 与恢复",
        "keywords": ["checkpoint", "恢复", "断点"],
        "content": "Checkpoint 保存图状态；外部写操作仍要通过幂等键防止重复执行。",
    },
    {
        "title": "Agent 循环",
        "keywords": ["agent", "循环", "tool"],
        "content": "Agent 在模型决策、工具执行、观察结果和再次决策之间循环。",
    },
    {
        "title": "RAG",
        "keywords": ["rag", "检索", "向量"],
        "content": "RAG 先检索外部资料，再把相关原文交给模型生成有依据的回答。",
    },
]

@tool
def search_notes(query: str, limit: int = 3) -> list[dict[str, str]]:
    """按关键词搜索技术笔记；回答笔记内容前应先调用此工具。

    Args:
        query: 用户想查询的技术概念或问题。
        limit: 最多返回多少条结果，默认返回 3 条。
    """
    normalized_query = query.lower()

    # 根据命中的关键词数量排序；这里只是便于理解的本地示例。
    ranked_notes = sorted(
        NOTES,
        key=lambda note: sum(
            keyword.lower() in normalized_query
            for keyword in note["keywords"]
        ),
        reverse=True,
    )

    # 只返回回答所需字段，避免把无关内部数据塞进模型 Context。
    return [
        {"title": note["title"], "content": note["content"]}
        for note in ranked_notes[:limit]
    ]

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

@tool 会利用函数名、类型注解和 docstring 生成模型可见的工具说明：

代码部分	模型或程序怎样使用
`search_notes`	默认成为 Tool 名称
`query: str`、`limit: int`	生成输入 Schema 并校验参数
docstring	告诉模型何时调用、参数是什么意思
返回值	作为 Tool observation 交还模型

工具描述太模糊时，模型可能不知道何时调用；Schema 只能保证参数形状，权限、业务规则和副作用控制仍由宿主应用负责。

# 再组装 Agent

# 文件位置：app/agent.py
import os

from dotenv import load_dotenv
from langchain.agents import create_agent
from langchain_openai import ChatOpenAI

from app.tools import search_notes

# 本地开发时读取项目根目录的 .env。
load_dotenv()

# 模型名来自可信配置，而不是来自用户消息。
model_name = os.getenv("OPENAI_MODEL")
if not model_name:
    raise RuntimeError("缺少 OPENAI_MODEL，请先配置 .env")

# ChatOpenAI 只负责模型调用；Agent 循环由 create_agent 组装。
model = ChatOpenAI(model=model_name)

# System Prompt 定义稳定职责、工具规则和证据边界。
SYSTEM_PROMPT = """
你是技术笔记问答助手。

- 回答笔记相关问题前，先调用 search_notes 查找依据。
- 只根据工具返回的笔记内容回答，不要虚构来源。
- 如果证据不足，直接说明缺少什么信息。
- 回答使用简洁中文，并在末尾列出引用的笔记标题。
""".strip()

# create_agent 返回一个可 invoke、stream 的已编译图。
agent = create_agent(
    model=model,                 # 负责推理和生成 Tool Call 的模型
    tools=[search_notes],        # 本次 Agent 可以发现和调用的工具集合
    system_prompt=SYSTEM_PROMPT, # 每次模型调用都会带上的稳定规则
)

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

# 最后调用 Agent

# 文件位置：run.py
from app.agent import agent

# messages 是 Agent 的默认状态字段；这里加入一条用户消息。
result = agent.invoke(
    {
        "messages": [
            {
                "role": "user",
                "content": "为什么有 Checkpoint 还要做幂等？",
            }
        ]
    }
)

# 最后一条消息是 Agent 完成工具循环后生成的最终回答。
print(result["messages"][-1].content)

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17

运行：

# 从项目根目录启动，确保 app 包可以被正确导入。
python run.py

1
2

# 一次请求到底经历了什么

用户问题
  ↓
create_agent 组装消息、System Prompt 和 Tool 描述
  ↓
模型判断需要查笔记，并生成 search_notes 的 Tool Call
  ↓
Runtime 校验参数并执行 search_notes
  ↓
搜索结果作为 ToolMessage 写回 messages
  ↓
模型读取搜索结果并生成最终回答
  ↓
invoke 返回包含完整消息轨迹的 State

1
2
3
4
5
6
7
8
9
10
11
12
13

Agent 的关键不是调用了一次模型，而是形成了 “模型决策 → 工具行动 → 观察结果 → 再决策” 的循环。

# 3. 模型、Prompt、Messages 与输出

# 四者不要混在一起

概念	负责什么	本例位置
Model	推理、选择工具、生成文本	`ChatOpenAI`
System Prompt	定义稳定职责和行为边界	`SYSTEM_PROMPT`
Messages	保存用户、模型和 Tool 的本轮轨迹	`result["messages"]`
Structured Output	把最终结果校验成程序可消费的结构	`response_format`

Prompt 告诉模型字段的业务含义，Schema 约束字段和类型，两者不能互相替代。

# 增加结构化输出

# 文件位置：app/schemas.py
from pydantic import BaseModel, Field

class NoteAnswer(BaseModel):
    """笔记问答助手最终返回的数据结构。"""

    answer: str = Field(description="给用户的中文回答")
    sources: list[str] = Field(description="实际引用的笔记标题")
    evidence_sufficient: bool = Field(description="当前证据是否足以回答")

1
2
3
4
5
6
7
8
9

# 文件位置：app/structured_agent.py
from langchain.agents import create_agent

from app.agent import SYSTEM_PROMPT, model
from app.schemas import NoteAnswer
from app.tools import search_notes

# 直接传入 Pydantic 类型时，LangChain 会按模型能力选择结构化输出策略。
structured_agent = create_agent(
    model=model,
    tools=[search_notes],
    system_prompt=SYSTEM_PROMPT,
    response_format=NoteAnswer, # 约束最终结果，不约束每条中间消息
)

result = structured_agent.invoke(
    {"messages": [{"role": "user", "content": "解释 Agent 循环"}]}
)

# 校验后的 Pydantic 对象保存在 structured_response 中。
answer: NoteAnswer = result["structured_response"]
print(answer.model_dump())

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22

Structured Output 能保证什么

它能保证返回字段可解析、类型符合 Schema；不能证明内容真实、来源有效或任务真的完成。事实正确性仍要靠检索证据、确定性校验和评测。

# 增加流式输出

invoke 等待最终结果；stream 可以把 Agent 每一步的进度或模型 Token 逐段交给前端。

# 文件位置：stream.py
from app.agent import agent

for chunk in agent.stream(
    {"messages": [{"role": "user", "content": "解释 RAG"}]},
    stream_mode="updates", # 每完成一个模型或 Tool 步骤就返回一次状态更新
    version="v2",          # 使用统一的事件结构
):
    if chunk["type"] != "updates":
        continue

    # data 的键通常是当前执行的图节点，例如 model 或 tools。
    for step_name, update in chunk["data"].items():
        latest_message = update["messages"][-1]
        print(step_name, latest_message.content_blocks)

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15

常用流模式：

模式	返回什么	前端用途
`updates`	每个 Agent 步骤的状态更新	展示 “正在选工具”、“检索完成”
`messages`	模型逐步生成的 Token 与元数据	打字机式回答
`custom`	Tool 或节点主动发送的自定义事件	展示处理进度

前端应该消费类型化事件，不要从自然语言中猜测当前状态。

# 4. Tools 工程化：模型能请求，不代表可以直接执行

一次 Tool 调用有三个责任主体：

模型：根据 Tool 描述提出名称和参数
  ↓
Agent Runtime：校验 Schema、调度执行、处理结果或错误
  ↓
领域服务：根据真实身份做授权、事务、幂等和审计

1
2
3
4
5

可靠 Tool 应满足：

一次只做一类明确动作；
参数 Schema 严格，描述说明何时用和何时不用；
身份从可信 Runtime Context 获取，不能相信模型传入的 user_id；
只返回后续决策需要的数据，避免泄露或撑大 Context；
写操作支持审批、幂等和审计；
错误要让上层知道是可重试、需改参数还是应立即停止。

Tool 还可以读取运行时的 State、Context 和 Store。三者区别是：

数据	生命周期	示例
State	同一 thread 内不断变化	messages、重试次数、当前草稿
Runtime Context	本次调用的可信只读依赖	登录用户、租户、数据库连接
Store	跨 thread 复用的长期信息	用户确认的偏好、长期目标

# 5. 加入 RAG：让 Agent 能搜索真实知识库

前面的 search_notes 只是关键词 Demo。真实 RAG 通常分为离线和在线两段：

离线：文档 → 清洗与切分 → Embedding → 写入向量库
在线：问题 → Embedding → 相似度检索 → 返回原文片段 → 模型回答

1
2

下面把内存笔记替换成最小向量检索工具：

# 文件位置：app/rag_tool.py
import os

from langchain.tools import tool
from langchain_core.documents import Document
from langchain_core.vectorstores import InMemoryVectorStore
from langchain_openai import OpenAIEmbeddings
from langchain_text_splitters import RecursiveCharacterTextSplitter

# 示例原文；生产项目应从文件、数据库或对象存储加载。
documents = [
    Document(
        page_content="Checkpoint 保存图状态，但外部写操作仍需幂等键。",
        metadata={"title": "Checkpoint 与恢复", "path": "/notes/checkpoint"},
    ),
    Document(
        page_content="Agent 通过模型决策、工具执行、观察和再决策形成循环。",
        metadata={"title": "Agent 循环", "path": "/notes/agent-loop"},
    ),
]

# 先按自然边界切分；chunk 太小会丢上下文，太大会混入噪声。
splitter = RecursiveCharacterTextSplitter(
    chunk_size=500,
    chunk_overlap=80,
)
chunks = splitter.split_documents(documents)

# 文档和查询必须使用兼容的 Embedding 模型与向量维度。
embeddings = OpenAIEmbeddings(
    model=os.getenv("OPENAI_EMBEDDING_MODEL", "text-embedding-3-small")
)

# 这里只使用进程内向量库；重启后会丢失，不能直接用于生产。
vector_store = InMemoryVectorStore.from_documents(
    documents=chunks,
    embedding=embeddings,
)

@tool
def search_knowledge(query: str, limit: int = 4) -> list[dict[str, str]]:
    """从技术知识库检索与问题语义相关的原文片段。"""
    matched_documents = vector_store.similarity_search(query, k=limit)

    # 来源字段由程序从 metadata 读取，不能让模型自己编造链接。
    return [
        {
            "content": document.page_content,
            "title": str(document.metadata["title"]),
            "path": str(document.metadata["path"]),
        }
        for document in matched_documents
    ]

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

把 create_agent 的 tools=[search_notes] 改成 tools=[search_knowledge]，模型便可以按需进行语义检索。

普通 RAG 与 Agentic RAG

普通 RAG 的检索步骤由代码固定执行；Agentic RAG 则让 Agent 根据当前问题和已有结果决定是否检索、改写查询、换数据源或继续补查。Agentic RAG 通常只是一个 Agent 多轮检索，不等于多个 Agent。

向量相似只表示语义上可能相关，不代表资料一定正确、完整或有访问权限。生产系统还要做 metadata 权限过滤、重排、引用校验和检索评测；也不应在每次进程启动时重新计算全部 Embedding。

# 6. Memory 与状态：让同一条会话可以继续

# 短期记忆来自 Checkpointer

没有 Checkpointer 时，每次 invoke 都只看到本次传入的 State。编译图时加入 Checkpointer，并在调用时提供 thread_id，才能恢复同一条 thread 的状态。

# 文件位置：app/memory_graph.py
import os

from dotenv import load_dotenv
from langchain_openai import ChatOpenAI
from langgraph.checkpoint.memory import InMemorySaver
from langgraph.graph import END, START, MessagesState, StateGraph

# 独立运行这个文件时也需要加载项目根目录的 .env。
load_dotenv()
model = ChatOpenAI(model=os.environ["OPENAI_MODEL"])

def call_model(state: MessagesState) -> dict:
    """读取当前消息状态，并追加一条模型回复。"""
    response = model.invoke(state["messages"])
    return {"messages": [response]}

builder = StateGraph(MessagesState)
builder.add_node("call_model", call_model) # 节点名称会出现在 Trace 中
builder.add_edge(START, "call_model")      # 用户输入先进入模型节点
builder.add_edge("call_model", END)        # 模型回复后结束本轮

# InMemorySaver 只适合本地演示；生产环境应换成数据库实现。
graph = builder.compile(checkpointer=InMemorySaver())

# thread_id 是查找哪组 Checkpoint 的游标，不等于用户身份。
config = {"configurable": {"thread_id": "interview-thread-001"}}

graph.invoke(
    {"messages": [{"role": "user", "content": "我叫小林"}]},
    config=config,
)
result = graph.invoke(
    {"messages": [{"role": "user", "content": "我叫什么？"}]},
    config=config,
)

print(result["messages"][-1].content)

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

Checkpoint 保存的是图状态，不保证外部副作用只执行一次。如果节点发完邮件、还没保存 Checkpoint 就崩溃，恢复后仍可能再次发邮件，所以写操作还要配合幂等键和事务。

# 五类信息不要混在一起

类型	作用	生命周期
消息历史	保存用户、模型和 Tool 消息	当前 thread，按需裁剪
工作记忆	保存本轮目标、临时结论和未完成步骤	当前任务结束前
长期记忆	保存未来仍会复用的用户事实或偏好	跨 thread，可更新和删除
Workflow State	保存阶段、产物、重试和审批状态	从任务创建到归档
Checkpoint	保存某一步可恢复的完整状态快照	按恢复和审计策略保留

Memory 保存以后可能有用的信息，State 说明任务现在走到哪里，Checkpoint 保存从哪里恢复。

# 7. 复杂任务为什么要用 LangGraph

create_agent 已经适合标准的 “模型选择 Tool 并循环直到回答” 场景。出现下面需求时，才值得直接编写 StateGraph：

固定校验必须在模型前后执行；
有明确的条件分支、并行步骤或循环；
任务运行很久，需要失败恢复；
高风险动作必须人工审批；
多个 Agent 或子流程需要清晰交接；
需要查看历史状态、回放或从旧状态分叉。

# State、Node、Edge 和 Reducer

概念	通俗理解	设计时问什么
State	整个流程共用的任务单	后续步骤真正需要哪些事实？
Node	处理任务单的一个步骤	能否独立重试和观测？
Edge	步骤之间的连线	下一步固定还是要判断？
Reducer	多次更新同一字段时的合并规则	追加、覆盖还是自定义合并？
START / END	流程入口与结束	什么条件才算真正完成？

# 写一个可暂停、可恢复的审核图

# 文件位置：app/review_graph.py
import os
from typing import Literal, TypedDict

from dotenv import load_dotenv
from langchain_openai import ChatOpenAI
from langgraph.checkpoint.memory import InMemorySaver
from langgraph.graph import END, START, StateGraph
from langgraph.types import Command, interrupt

class ReviewState(TypedDict):
    """审核流程在节点之间共享的状态。"""

    question: str            # 用户原始问题
    draft: str               # 模型生成、等待审核的草稿
    status: str              # pending、approved 或 rejected

# 独立运行这个文件时也需要加载项目根目录的 .env。
load_dotenv()
model = ChatOpenAI(model=os.environ["OPENAI_MODEL"])

def generate_draft(state: ReviewState) -> dict:
    """根据问题生成草稿，只更新 draft 和 status。"""
    response = model.invoke(
        f"用简洁中文回答下面问题：{state['question']}"
    )
    return {"draft": str(response.content), "status": "pending"}

def review_draft(
    state: ReviewState,
) -> Command[Literal["finish", "cancel"]]:
    """暂停图并等待外部用户审核，然后决定下一条边。"""
    approved = interrupt(
        {
            "question": "是否发布这份回答？",
            "draft": state["draft"],
        }
    )

    # resume 传入 True 时进入 finish，否则进入 cancel。
    if approved:
        return Command(update={"status": "approved"}, goto="finish")
    return Command(update={"status": "rejected"}, goto="cancel")

def finish(state: ReviewState) -> dict:
    """批准后的确定性收尾节点。"""
    return {"status": "approved"}

def cancel(state: ReviewState) -> dict:
    """拒绝后的确定性收尾节点。"""
    return {"status": "rejected"}

builder = StateGraph(ReviewState)
builder.add_node("generate_draft", generate_draft)
builder.add_node("review_draft", review_draft)
builder.add_node("finish", finish)
builder.add_node("cancel", cancel)

builder.add_edge(START, "generate_draft")
builder.add_edge("generate_draft", "review_draft")
builder.add_edge("finish", END)
builder.add_edge("cancel", END)

# Interrupt 依赖 Checkpointer 保存暂停位置。
review_graph = builder.compile(checkpointer=InMemorySaver())

# 首次调用会在 review_draft 的 interrupt 处暂停。
config = {"configurable": {"thread_id": "review-001"}}
paused = review_graph.invoke(
    {"question": "什么是 Agent？", "draft": "", "status": "pending"},
    config=config,
)
print(paused["__interrupt__"])

# 使用相同 thread_id 恢复；True 表示批准当前草稿。
resumed = review_graph.invoke(Command(resume=True), config=config)
print(resumed["status"])

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
70
71
72
73
74
75
76
77

这段代码的运行路径是：

START
  └─ generate_draft
      └─ review_draft
          ├─ interrupt：保存状态并等待用户
          ├─ resume=True  → finish → END
          └─ resume=False → cancel → END

1
2
3
4
5
6

恢复时包含 interrupt() 的节点可能重新执行，所以中断之前的副作用必须幂等。生产环境也不能使用 InMemorySaver，应替换为 PostgreSQL 等持久化 Checkpointer。

# Durable Execution 不是 Exactly-once

Durable Execution：进程中断后，任务仍能从已保存状态继续；
Exactly-once：同一个业务副作用即使发生重试，也只生效一次。

LangGraph 解决前者，后者仍需要业务数据库、幂等键、事务或去重记录配合。

# Subgraph 解决什么

Subgraph 是封装好的局部状态机，可以作为父图的一个节点。它适合复用复杂子流程或隔离不同 Agent 的状态，不只是为了拆文件。多 Agent 默认应避免让子 Agent 无限制积累历史，否则旧任务容易污染新任务。

# 8. LangSmith：从 Trace 走到评测闭环

本地 print 只能看到最终结果，真实 Agent 还要知道模型为什么选了某个 Tool、哪一步慢、修改 Prompt 后是否真的变好。LangSmith 主要解决这些问题。

概念	含义
Project	同一应用或环境的一组 Trace
Trace	一次用户请求的完整调用链
Run	Trace 中的一次模型、Tool、Retriever 或函数执行
Thread	同一多轮会话关联的多条 Trace
Dataset	可重复运行的评测输入和参考信息
Experiment	某个应用版本在 Dataset 上的运行与得分

# 打开 Tracing

# 文件位置：.env
# 打开 LangSmith 自动追踪。
LANGSMITH_TRACING=true

# 使用你自己的 LangSmith API Key，不要提交到仓库。
LANGSMITH_API_KEY=replace_with_your_langsmith_key

# 把不同应用或环境的 Trace 分开管理。
LANGSMITH_PROJECT=interview-note-agent-dev

1
2
3
4
5
6
7
8
9

LangChain Agent 运行后便可在 LangSmith 中查看模型、Tool 和延迟轨迹。敏感项目要先评估脱敏、采样、数据保留和合规要求。

# 一次评测需要哪三样东西

Dataset：固定测试题与可选参考答案
  └─ Target：要测试的 Agent、Graph 或函数
      └─ Evaluator：根据事实、工具、安全等规则打分
          └─ Experiment：保存本次版本的全部结果

1
2
3
4

正确的改进循环是：

生产失败样例
  └─ 脱敏并加入 Dataset
      └─ 修改 Prompt、Tool、模型或图
          └─ 运行相同 Experiment
              └─ 比较质量、延迟、费用和安全指标

1
2
3
4
5

不要只看平均分，还要查看具体失败 Trace。一次整体分数上升，可能掩盖某一类关键问题退化。

# 9. 部署与生态：部署的不是一段 Prompt

开发完成后有两条常见路线：

路线	怎样部署	适合什么情况
自建 API	在 FastAPI、Django 等服务中调用 `agent.invoke` 或 `agent.stream`	已有后端体系，需要完全控制认证、队列和数据库
Agent Server	把 Graph 部署到 LangSmith Deployment 或自托管 Agent Server	需要内置 Thread、Run、持久化、队列、流式 API 和 Studio

Agent Server 部署的核心不只是模型代码，还包括一个或多个 Graph、持久化数据库和任务队列。客户端可以通过 SDK 或 HTTP API 创建 Thread、启动 Run、订阅流式事件并恢复中断。

Web / App / Backend
  └─ SDK 或 HTTP 请求
      └─ Agent Server
          ├─ Assistant：某个 Graph 的可配置版本
          ├─ Thread：一条持续会话的状态容器
          ├─ Run：一次具体执行
          ├─ Persistence：Checkpoint 与 Store
          └─ Task Queue：调度长任务和并发执行

1
2
3
4
5
6
7
8

生态中还可能看到：

Studio：连接本地或远程 Graph，查看状态、运行轨迹和中断；
LangGraph SDK：从前端或后端调用 Agent Server；
MCP Endpoint：把 Agent Server 的能力按 MCP 方式提供给兼容 Host；
A2A Endpoint：让独立 Agent 之间按 Agent-to-Agent 协议协作；
普通 Web API：由自己的后端包装业务接口，仍然是最通用的接入方式。

MCP、A2A 和 Agent Server 解决的是不同问题，不是部署 LangChain Agent 的三个必选步骤。

# 10. 回头再看三者边界

组件	核心问题	主要能力	不负责什么
LangChain	怎样快速组装标准 Agent	Model、Messages、Tool、Middleware、Structured Output、`create_agent`	业务授权、事务和完整生产基础设施
LangGraph	怎样控制复杂、有状态的执行过程	State、Node、Edge、Reducer、Checkpoint、Interrupt、Subgraph	自动替你决定业务流程是否合理
LangSmith	怎样观察、评测和部署 Agent	Trace、Dataset、Experiment、Monitoring、Deployment	替代领域测试、安全边界和业务验收

create_agent 底层使用 LangGraph Runtime，因此简单 Agent 不需要再手写一层空图。只有当默认循环不能清楚表达业务控制流时，才下沉到 StateGraph。

# Middleware 放在哪里

Middleware 适合多个调用都会经过的横切处理：

模型路由、统一重试和缓存；
PII 脱敏、内容裁剪和摘要；
Tool 调用前后的错误转换；
统一调用预算和日志标签。

关键业务顺序不应隐藏在 Middleware 中。例如 “付款前必须先风控再审批” 应写成显式 Node 和 Edge，否则难以测试、观察和恢复。

# 11. 项目中怎样选择层级

需求	推荐起点	原因
单轮分类或结构化生成	Model + Structured Output	没有循环，不需要 Agent
标准工具型 Agent	LangChain `create_agent`	默认循环已经够用
固定 RAG 问答	2-step RAG Chain	路径稳定，延迟和成本更可控
按需多轮检索	`create_agent` + Retriever Tool	由模型决定何时继续检索
复杂分支、长任务或审批	LangGraph `StateGraph`	状态、路径和恢复点显式
调试、回归和监控	LangSmith 或等价平台	把运行轨迹和质量指标连接起来

面试中不要只说使用了哪些库，还要说明为什么选择这一层，以及哪些责任仍留在宿主应用中。

# 12. 一个可用于简历的纵向切片

可以用同一个笔记助手证明完整能力：

用 create_agent 组装模型和检索 Tool；
把 Markdown 笔记切分、向量化并写入向量库；
检索时先按真实用户权限过滤，再做向量召回和重排；
用 Structured Output 返回答案、来源和证据状态；
用 LangGraph 固定 “检索 → 生成 → 评测 → 审批 → 保存” 流程；
用持久 Checkpointer 和 thread_id 支持暂停与恢复；
用 LangSmith Trace 定位失败，并用 Dataset 比较 Prompt 或模型版本；
通过自建 API 或 Agent Server 向前端提供流式事件。

项目表达可以这样组织：业务问题 → 为什么需要 Agent → 为什么选择 LangChain / LangGraph → 状态与工具怎样设计 → 如何保证安全和恢复 → 怎样用评测证明有效。

# 高频面试题与回答

回答顺序：先给一句结论，再解释运行过程，最后补一个项目例子或工程边界。

1. 2 分钟回答：LangChain、LangGraph、LangSmith 怎样配合参考答案

我会把三者看成组装、运行和改进三个层次。LangChain 提供 Model、Tool、Messages、Middleware 和 Structured Output 等常用组件，create_agent 可以快速搭建标准的模型与工具循环；它底层使用 LangGraph Runtime。遇到复杂分支、长任务、持久状态或人工审批时，我会直接使用 LangGraph，把 State、Node、Edge 和停止条件写清楚。运行后再用 LangSmith 查看 Trace、沉淀 Dataset，并通过 Experiment 比较 Prompt、模型或工具版本。真实身份、权限、事务和幂等仍由宿主应用负责。

2. 一次 LangChain Agent 请求经历了什么？参考答案

宿主应用先准备用户消息、System Prompt、可用 Tool 和运行上下文，模型随后决定直接回答还是提出 Tool Call。Runtime 校验参数并执行真实工具，把结果作为 ToolMessage 写回 State，模型再根据新状态继续决策，直到生成最终答案或触发停止条件。模型只提出工具请求，真正的权限校验和执行发生在宿主应用。

3. create_agent 和 StateGraph 怎样选择？参考答案

如果任务只是标准的 “模型选工具 → 执行 → 观察结果 → 继续或结束”，优先用 create_agent。当流程包含固定校验、复杂分支、并行、审批、长任务恢复或多个子流程时，再直接写 StateGraph。判断标准不是哪个更高级，而是默认循环能不能清楚表达业务控制流。

4. 普通 RAG 和 Agentic RAG 有什么区别？参考答案

普通 RAG 通常由代码固定执行一次检索再生成，路径可预测、成本较低；Agentic RAG 把检索能力做成 Tool，让 Agent 根据问题和已有结果决定是否检索、改写查询、切换数据源或继续补查。后者适合复杂问题，但会增加延迟、费用和失败路径。Agentic RAG 不等于 Multi-Agent，它也可以只是一个 Agent 多轮检索。

5. State、Runtime Context、Store 和 Checkpoint 有什么区别？参考答案

State 保存同一条任务线程中不断变化的数据；Runtime Context 保存本次调用由宿主应用注入的可信依赖；Store 保存跨线程仍要复用的长期信息；Checkpoint 则是某一步完整 State 的可恢复快照。可以记成：State 管当前任务，Context 管本次运行环境，Store 管长期信息，Checkpoint 管从哪里恢复。

6. Checkpoint 为什么不能保证写操作只执行一次？参考答案

因为 Checkpoint 和外部系统不一定在同一个事务中提交。节点可能已经发出邮件或写入数据，但在保存 Checkpoint 前崩溃；恢复后节点重新执行，就会产生重复副作用。因此外部写操作仍要使用幂等键、事务、去重记录和必要审批。

7. LangSmith 是运行 Agent 的必需组件吗？参考答案

不是。LangChain 和 LangGraph 可以独立运行，LangSmith 是可选的追踪、评测、监控和部署平台。使用它可以更快查看 Trace、管理 Dataset 和比较 Experiment，但团队仍要根据数据敏感性、合规要求和自建成本决定是否接入。

8. LangChain Agent 怎样部署？参考答案

常见有两种方式：一种是在已有 FastAPI 或其他后端中调用 Agent，自行负责认证、持久化、队列和流式接口；另一种是部署到 Agent Server，使用内置的 Assistant、Thread、Run、持久化和任务队列能力。无论选择哪种，生产部署都不是只上传 Prompt，还要同时处理密钥、权限、状态、并发、评测和监控。

# 接下来学什么

下一篇学习完整项目与系统设计，重点把需求、架构、工具、知识、状态、安全、评测、前端和框架选型整合成一个可实现、可展示、可面试的完整项目。

# 参考资料

Mastra 核心原理完整项目与系统设计

首页

Choose mode

LangChain Agents 开发：从 create_agent 到部署

# LangChain Agents 开发：从 create_agent 到部署

# 先记住一句话

# 先看完整开发路径

# 1. 项目启动

# 安装与目录

# 2. 从零写一个可运行的 Agent

# 先定义 Tool

# 再组装 Agent

# 最后调用 Agent

# 一次请求到底经历了什么

# 3. 模型、Prompt、Messages 与输出

# 四者不要混在一起

# 增加结构化输出

# 增加流式输出

# 4. Tools 工程化：模型能请求，不代表可以直接执行

# 5. 加入 RAG：让 Agent 能搜索真实知识库

# 6. Memory 与状态：让同一条会话可以继续

# 短期记忆来自 Checkpointer

# 五类信息不要混在一起

# 7. 复杂任务为什么要用 LangGraph

# State、Node、Edge 和 Reducer

# 写一个可暂停、可恢复的审核图

# Durable Execution 不是 Exactly-once

# Subgraph 解决什么

# 8. LangSmith：从 Trace 走到评测闭环

# 打开 Tracing

# 一次评测需要哪三样东西

# 9. 部署与生态：部署的不是一段 Prompt

# 10. 回头再看三者边界

# Middleware 放在哪里

# 11. 项目中怎样选择层级

# 12. 一个可用于简历的纵向切片

# 高频面试题与回答

# 接下来学什么

# 参考资料