LangChain 与 LangGraph 技术栈：核心抽象、工作流编排与生产实践

┌─────────────────────────────────────────────────┐
│              业务应用层 (Your App)               │
├─────────────────────────────────────────────────┤
│         LangGraph (工作流编排 / Agent Loop)      │
├─────────────────────────────────────────────────┤
│    LangChain (Chain / Retrieval / Tool 范式)    │
├─────────────────────────────────────────────────┤
│      langchain-core (Runnable / LCEL 协议)      │
├─────────────────────────────────────────────────┤
│  LangServe (部署)  │  LangSmith (可观测性)      │
├─────────────────────────────────────────────────┤
│     LLM Provider / 向量数据库 / 工具 API        │
└─────────────────────────────────────────────────┘

from langchain_openai import ChatOpenAI
from langchain_core.messages import HumanMessage, SystemMessage

llm = ChatOpenAI(model="gpt-4o", temperature=0)

messages = [
    SystemMessage(content="你是一位资深的 Python 架构师"),
    HumanMessage(content="请解释 Liskov 替换原则，并给出 Python 示例")
]

response = llm.invoke(messages)
print(response.content)

for chunk in llm.stream(messages):
    print(chunk.content, end="", flush=True)

from langchain_core.prompts import ChatPromptTemplate, FewShotPromptTemplate
from langchain_core.prompts import PromptTemplate

# 基础模板
prompt = ChatPromptTemplate.from_messages([
    ("system", "你是一位{role}，擅长{domain}领域"),
    ("user", "{question}")
])

formatted = prompt.invoke({
    "role": "技术顾问",
    "domain": "分布式系统",
    "question": "如何设计一个高可用的消息队列？"
})

from langchain_core.prompts import FewShotChatMessagePromptTemplate

examples = [
    {"input": "这个产品太棒了", "output": "positive"},
    {"input": "完全不推荐", "output": "negative"},
    {"input": "还行吧", "output": "neutral"},
]

example_prompt = ChatPromptTemplate.from_messages([
    ("user", "{input}"),
    ("assistant", "{output}")
])

few_shot_prompt = FewShotChatMessagePromptTemplate(
    example_prompt=example_prompt,
    examples=examples
)

final_prompt = ChatPromptTemplate.from_messages([
    ("system", "你是一位情感分析专家。请将用户评论分类为 positive/negative/neutral。"),
    few_shot_prompt,
    ("user", "{input}")
])

from langchain_openai import ChatOpenAI
from langchain_core.prompts import ChatPromptTemplate
from langchain_core.output_parsers import StrOutputParser

prompt = ChatPromptTemplate.from_template(
    "请用一段话解释以下概念：{concept}"
)
llm = ChatOpenAI(model="gpt-4o-mini", temperature=0.7)
parser = StrOutputParser()

chain = prompt | llm | parser          # LCEL：用 | 把组件编成管道
result = chain.invoke({"concept": "幂等性"})

from langchain.chains import create_retrieval_chain
from langchain.chains.combine_documents import create_stuff_documents_chain

qa_chain = create_stuff_documents_chain(llm, RAG_PROMPT)  # 检索文档注入 {context}
pipeline = create_retrieval_chain(
    vectorstore.as_retriever(search_kwargs={"k": 4}), qa_chain)

from langgraph.graph import StateGraph, MessagesState, START, END
from langgraph.prebuilt import ToolNode

def call_model(state: MessagesState):
    return {"messages": [llm_with_tools.invoke(state["messages"])]}

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

graph = StateGraph(MessagesState)
graph.add_node("agent", call_model)
graph.add_node("tools", ToolNode(tools))
graph.add_edge(START, "agent")
graph.add_conditional_edges("agent", should_continue,
                            {"tools": "tools", END: END})
graph.add_edge("tools", "agent")

app = graph.compile(checkpointer=SqliteSaver.from_conn_string("ckpt.db"))