LangChain、LangGraph 与 LangSmith：从原型到生产

做大模型应用时，LangChain 是很多人最先接触的框架。它把 Prompt 模板、模型调用、工具链、向量检索封装成可组合的模块，几行代码就能跑出一个原型。但当流程变复杂，单次调用变成多轮循环、条件分支、状态流转时，就需要 LangGraph。而要把这套东西部署到生产环境，又离不开 LangSmith 的追踪与评测。

这篇文章用实际代码串起三者：LangChain 搭基础、LangGraph 做编排、LangSmith 做观测。

一、LangChain：把 LLM 调用工程化

LangChain 的核心抽象是 Runnable 与 Chain。一个 Prompt 模板 + 一个模型 + 一个输出解析器，就可以组成一条链。

# pip install langchain langchain-openai
from langchain_core.prompts import ChatPromptTemplate
from langchain_openai import ChatOpenAI
from langchain_core.output_parsers import StrOutputParser

prompt = ChatPromptTemplate.from_messages([
    ("system", "你是一位资深 Python 工程师，用中文回答。"),
    ("human", "请解释 {topic}，并给出一个代码示例。")
])

model = ChatOpenAI(model="gpt-4o-mini", temperature=0.2)
chain = prompt | model | StrOutputParser()

print(chain.invoke({"topic": "Python 装饰器"}))

这里 prompt | model | parser 是 LCEL（LangChain Expression Language），用管道符把多个 Runnables 串起来。它的好处是：

每一环都是可替换的，换模型、换提示、换解析器都很方便；
天然支持 .batch()、.stream()、.astream()；
中间状态可观测，便于接入 LangSmith。

1.1 接入 RAG：Retrieval Chain

把向量检索加进链里，就是最常见的 RAG 流程：

from langchain_community.vectorstores import FAISS
from langchain_openai import OpenAIEmbeddings
from langchain_core.runnables import RunnablePassthrough

vectorstore = FAISS.from_texts(
    ["LangChain 是 LLM 应用框架", "LangGraph 用于状态图编排"],
    embedding=OpenAIEmbeddings()
)
retriever = vectorstore.as_retriever(search_kwargs={"k": 3})

template = """基于以下上下文回答问题：
{context}

问题：{question}
"""
prompt = ChatPromptTemplate.from_template(template)

rag_chain = (
    {"context": retriever, "question": RunnablePassthrough()}
    | prompt
    | model
    | StrOutputParser()
)

print(rag_chain.invoke("LangChain 和 LangGraph 有什么关系？"))

二、LangGraph：当 Chain 变成图

LangChain 的链是线性的，适合单次检索 + 生成。但 Agent 需要循环：观察 → 思考 → 行动 → 再观察。LangGraph 用 StateGraph 把状态流转显式化。

最简单的 Agent 图包含三个节点：

agent：LLM 决定调用哪个工具；
action：执行工具；
should_continue：判断是继续循环还是返回结果。

from typing import TypedDict, Annotated, Sequence
import operator
from langchain_core.messages import BaseMessage, HumanMessage, ToolMessage
from langgraph.graph import StateGraph, END
from langchain_openai import ChatOpenAI
from langchain.tools import tool

@tool
def search_web(query: str) -> str:
    """模拟网络搜索工具"""
    return f"关于 '{query}' 的搜索结果：..."

class AgentState(TypedDict):
    messages: Annotated[Sequence[BaseMessage], operator.add]

tools = [search_web]
model = ChatOpenAI(model="gpt-4o-mini").bind_tools(tools)

def agent_node(state: AgentState):
    return {"messages": [model.invoke(state["messages"])]}

def action_node(state: AgentState):
    last_message = state["messages"][-1]
    tool_calls = last_message.tool_calls
    results = []
    for tc in tool_calls:
        for t in tools:
            if t.name == tc["name"]:
                result = t.invoke(tc["args"])
                results.append(ToolMessage(content=str(result), tool_call_id=tc["id"]))
    return {"messages": results}

def should_continue(state: AgentState):
    last = state["messages"][-1]
    return "action" if last.tool_calls else END

workflow = StateGraph(AgentState)
workflow.add_node("agent", agent_node)
workflow.add_node("action", action_node)
workflow.set_entry_point("agent")
workflow.add_conditional_edges("agent", should_continue)
workflow.add_edge("action", "agent")

app = workflow.compile()

for event in app.stream({"messages": [HumanMessage(content="今天北京天气怎么样？")]}):
    print(event)

LangGraph 的价值在于：

状态显式：所有节点共享一个 State，调试时能看到完整上下文；
循环与分支：通过 add_conditional_edges 实现复杂控制流；
人机协同：可以在节点中插入中断（interrupt），等用户确认再继续。

三、LangSmith：可观测性与评测

把链或图抛到生产环境后，必须回答三个问题：调用链里哪一步最慢？哪个输入产生了幻觉？模型升级后效果有没有退化？LangSmith 就是解决这些的。

3.1 自动追踪

只要设置两个环境变量，LangChain 的所有调用自动上报：

export LANGCHAIN_TRACING_V2=true
export LANGCHAIN_API_KEY="ls-..."
export LANGCHAIN_PROJECT="judeai-demo"

之后每次 chain.invoke() 或 app.stream()，LangSmith 都会记录完整的输入、输出、延迟、Token 消耗，并以树状结构展示。

3.2 数据集与评测

LangSmith 支持创建测试集，然后批量跑评测。下面是一个简单示例：

from langsmith import Client
from langsmith.evaluation import evaluate

client = Client()

dataset = client.create_dataset(
    dataset_name="rag_qa",
    description="RAG 问答评测集"
)
client.create_examples(
    inputs=[
        {"question": "LangChain 是什么？"},
        {"question": "LangGraph 解决什么问题？"}
    ],
    outputs=[
        {"answer": "一个 LLM 应用开发框架。"},
        {"answer": "把 LLM 工作流建模为状态图，支持循环和分支。"}
    ],
    dataset_id=dataset.id
)

def predict(example):
    return {"answer": rag_chain.invoke(example["question"])}

def correctness(run, example):
    # 实际生产中会调用 LLM-as-judge 或人工标注
    return {"score": 1.0 if example.outputs["answer"] in run.outputs["answer"] else 0.0}

results = evaluate(
    predict,
    data="rag_qa",
    evaluators=[correctness],
    experiment_prefix="rag-exp"
)

四、三者怎么配合

一个典型的生产链路是：

用 LangChain 把 Prompt、模型、工具、向量库串成可复用的 Runnables；
用 LangGraph 把这些 Runnables 组织成带状态、循环、分支的 Agent 工作流；
用 LangSmith 追踪线上调用、收集 bad case、构建评测集、回归验证新模型。

LangChain 是零件，LangGraph 是流水线，LangSmith 是质检与监控系统。单用任何一个都能跑，但三者组合才能把 LLM 应用从 demo 推到生产。

五、避坑建议

不要过度抽象：简单场景直接用 OpenAI SDK 或 requests 调用模型，比硬套 LangChain 更轻量；
版本锁定：LangChain 生态迭代快，langchain-core、langchain-community、各 provider 包版本要对齐；
状态体积控制：LangGraph 的 State 会往返于各节点，消息历史太长时要做截断或摘要；
评测要自动化：上线前至少准备 50+ 条代表业务场景的用例，用 LLM-as-judge 或规则评分做回归。

小结

LangChain、LangGraph、LangSmith 分别解决 LLM 应用的组合、编排、观测三个问题。写 demo 时 LangChain 就够了；做 Agent 时用 LangGraph 把循环和分支显式化；要上线时 LangSmith 提供追踪、评测和数据飞轮。

建议边写代码边上 LangSmith 看 trace，很多隐藏问题（比如重复检索、工具误调用、Token 爆炸）在可视化链路里一目了然。

← 返回首页