LangChain框架全解析——架构、组件与实战

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

# 1. 定义提示模板
prompt = ChatPromptTemplate.from_template(
    "你是一位资深 {role}，请用通俗易懂的方式解释 {concept}"
)

# 2. 选择模型
llm = ChatOpenAI(model="gpt-4", temperature=0.7)

# 3. 定义输出解析器
parser = StrOutputParser()

# 4. 用管道符组合成链
chain = prompt | llm | parser

# 5. 调用
result = chain.invoke({"role": "物理学家", "concept": "量子纠缠"})
print(result)

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

# ChatModel（推荐，现代模型都是对话模型）
chat_model = ChatOpenAI(model="gpt-4", temperature=0)

# 方式一：直接传字符串
response = chat_model.invoke("什么是 RAG？")

# 方式二：传消息列表（更灵活，能设定 system 角色）
messages = [
    SystemMessage(content="你是一位耐心的 AI 老师，回答要简洁准确"),
    HumanMessage(content="什么是 RAG？"),
]
response = chat_model.invoke(messages)
print(response.content)
print(f"Token 用量: {response.usage_metadata}")

# 切换到 Anthropic Claude
from langchain_anthropic import ChatAnthropic
chat_model = ChatAnthropic(model="claude-sonnet-4-20250514")

# 切换到本地 Ollama
from langchain_ollama import ChatOllama
chat_model = ChatOllama(model="llama3")
# 业务代码完全不用改

from langchain_core.prompts import ChatPromptTemplate, PromptTemplate

# 字符串模板（用于 LLM）
string_prompt = PromptTemplate.from_template(
    "请把以下文本翻译成{language}：\n{text}"
)

# 聊天模板（用于 ChatModel，推荐）
chat_prompt = ChatPromptTemplate.from_messages([
    ("system", "你是一位专业的{role}，回答要{style}"),
    ("human", "{question}"),
])

# 渲染
formatted = chat_prompt.invoke({
    "role": "数据科学家",
    "style": "深入浅出",
    "question": "解释什么是梯度下降"
})
print(formatted)
# -> messages=[SystemMessage('你是一位专业的数据科学家...'), HumanMessage('解释什么是梯度下降')]

from langchain_core.example_selectors import SemanticSimilarityExampleSelector
from langchain_core.prompts import FewShotChatMessagePromptTemplate
from langchain_openai import OpenAIEmbeddings
from langchain_chroma import Chroma

# 准备示例库
examples = [
    {"input": "高兴", "output": "悲伤"},
    {"input": "高大", "output": "矮小"},
    {"input": "精力充沛", "output": "无精打采"},
    {"input": "简单", "output": "复杂"},
    {"input": "快", "output": "慢"},
]

# 语义相似度选择器：根据输入动态选择最相关的示例
example_selector = SemanticSimilarityExampleSelector.from_examples(
    examples, OpenAIEmbeddings(), Chroma(), k=2
)

# 构建 few-shot 提示
few_shot_prompt = FewShotChatMessagePromptTemplate(
    example_selector=example_selector,
    example_prompt=ChatPromptTemplate.from_messages([
        ("human", "{input}"),
        ("ai", "{output}"),
    ]),
)

# 组合成完整提示
final_prompt = ChatPromptTemplate.from_messages([
    ("system", "给出输入词的反义词"),
    few_shot_prompt,
    ("human", "{input}"),
])

chain = final_prompt | ChatOpenAI(model="gpt-4")
print(chain.invoke({"input": "热情"}).content)

from langchain_core.output_parsers import PydanticOutputParser
from pydantic import BaseModel, Field

# 定义期望的输出结构
class MovieReview(BaseModel):
    title: str = Field(description="电影标题")
    rating: int = Field(description="评分 1-10")
    summary: str = Field(description="一句话影评")
    pros: list[str] = Field(description="优点列表")
    cons: list[str] = Field(description="缺点列表")

# 创建解析器
parser = PydanticOutputParser(pydantic_object=MovieReview)

# 解析器会自动生成格式说明，注入到提示里
prompt = ChatPromptTemplate.from_messages([
    ("system", "你是一位影评人。请按指定格式输出影评。\n{format_instructions}"),
    ("human", "请评价电影：{movie}"),
]).partial(format_instructions=parser.get_format_instructions())

chain = prompt | ChatOpenAI(model="gpt-4", temperature=0) | parser

# 拿到的是结构化对象，不是字符串
review = chain.invoke({"movie": "盗梦空间"})
print(f"电影: {review.title}, 评分: {review.rating}/10")
print(f"优点: {review.pros}")

from langchain_community.document_loaders import (
    TextLoader, WebBaseLoader, PyPDFLoader, DirectoryLoader
)

# 加载单个文本文件
loader = TextLoader("./knowledge_base.txt")
docs = loader.load()

# 加载网页
web_loader = WebBaseLoader("https://example.com/article")
web_docs = web_loader.load()

# 加载 PDF
pdf_loader = PyPDFLoader("./report.pdf")
pdf_docs = pdf_loader.load()  # 每页一个 Document

# 批量加载目录下所有文件
dir_loader = DirectoryLoader("./docs/", glob="**/*.pdf",
                              loader_cls=PyPDFLoader)
all_docs = dir_loader.load()

print(f"共加载 {len(all_docs)} 个文档")
print(f"第一个文档元数据: {all_docs[0].metadata}")

from langchain_text_splitters import RecursiveCharacterTextSplitter

# 递归字符拆分器：优先按段落、然后按句子、最后按字符拆分
splitter = RecursiveCharacterTextSplitter(
    chunk_size=1000,      # 每块最大 1000 字符
    chunk_overlap=200,    # 块之间重叠 200 字符，保持上下文连贯
    separators=["\n\n", "\n", "。", "！", "？", "，", " ", ""],
)

chunks = splitter.split_documents(all_docs)
print(f"拆分前 {len(all_docs)} 个文档 → 拆分后 {len(chunks)} 个块")

from langchain_openai import OpenAIEmbeddings

embeddings = OpenAIEmbeddings(model="text-embedding-3-small")

# 嵌入单条文本
vector = embeddings.embed_query("LangChain 是一个 LLM 应用框架")
print(f"向量维度: {len(vector)}")

# 批量嵌入
texts = ["RAG 是检索增强生成", "Agent 能自主调用工具"]
vectors = embeddings.embed_documents(texts)

from langchain_chroma import Chroma

# 从文档块创建向量库
vectorstore = Chroma.from_documents(
    documents=chunks,
    embedding=OpenAIEmbeddings(),
    persist_directory="./chroma_db",  # 持久化到磁盘
)

# 相似度搜索
results = vectorstore.similarity_search(
    "LangChain 的核心组件有哪些？",
    k=3,  # 返回最相似的 3 个
)
for doc in results:
    print(doc.page_content[:100], "...")

# 带分数的搜索
results_with_score = vectorstore.similarity_search_with_score("什么是 RAG", k=3)
for doc, score in results_with_score:
    print(f"分数: {score:.4f} | {doc.page_content[:80]}...")

from langchain_core.retrievers import BaseRetriever

# 最简单：把向量库转成检索器
retriever = vectorstore.as_retriever(
    search_type="similarity",  # 也可用 mmr（最大边际相关性，去重）
    search_kwargs={"k": 4},
)

docs = retriever.invoke("LangChain 的 Memory 组件")
print(f"检索到 {len(docs)} 个相关文档")

# 自定义检索器
class CustomRetriever(BaseRetriever):
    def _get_relevant_documents(self, query):
        # 这里可以混用：向量检索 + 关键词检索 + 规则过滤
        return vectorstore.similarity_search(query, k=2)

from langchain.indexes import SQLRecordManager, index

# 记录管理器：追踪哪些文档已索引
record_manager = SQLRecordManager(
    namespace="chroma/my_docs",
    db_url="sqlite:///record_manager.db"
)
record_manager.create_schema()

# 执行索引（自动去重、增量更新）
result = index(
    chunks,
    record_manager,
    vectorstore,
    cleanup="incremental",  # 增量模式：新增的写入，删除的移除
    source_id_key="source",
)
print(f"新增: {result['num_added']}, 删除: {result['num_deleted']}, 更新: {result['num_updated']}, 跳过: {result['num_skipped']}")

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

# 检索 + 生成
template = """请根据以下上下文回答问题。如果上下文中没有相关信息，请说"我不知道"。

上下文：
{context}

问题：{question}
"""
prompt = ChatPromptTemplate.from_template(template)
llm = ChatOpenAI(model="gpt-4", temperature=0)

def format_docs(docs):
    return "\n\n".join(doc.page_content for doc in docs)

# 经典 RAG 链
rag_chain = (
    {"context": retriever | format_docs, "question": RunnablePassthrough()}
    | prompt
    | llm
    | StrOutputParser()
)

answer = rag_chain.invoke("LangChain 的 Memory 组件有什么作用？")
print(answer)

# 传统 LLMChain（已被 LCEL 替代，但理解原理有价值）
from langchain.chains import LLMChain

chain = LLMChain(
    llm=ChatOpenAI(model="gpt-4"),
    prompt=ChatPromptTemplate.from_template("讲一个关于{subject}的笑话"),
)
print(chain.invoke({"subject": "程序员"}))

# 顺序链：第一步总结，第二步翻译
from langchain.chains import SimpleSequentialChain

summary_chain = LLMChain(
    llm=ChatOpenAI(model="gpt-4"),
    prompt=ChatPromptTemplate.from_template("用一句话总结：{text}")
)
translate_chain = LLMChain(
    llm=ChatOpenAI(model="gpt-4"),
    prompt=ChatPromptTemplate.from_template("把以下内容翻译成英文：{text}")
)
overall_chain = SimpleSequentialChain(
    chains=[summary_chain, translate_chain], verbose=True
)
result = overall_chain.run("LangChain 是一个用于构建 LLM 应用的框架，提供六大核心组件...")
print(result)

from langchain.chains import ConversationChain
from langchain.memory import (
    ConversationBufferMemory,       # 完整对话历史
    ConversationBufferWindowMemory, # 只保留最近 N 轮
    ConversationTokenBufferMemory,  # 按 Token 数限制
    ConversationSummaryMemory,      # 摘要式记忆
)

# 1. 完整缓冲区记忆（简单但会撑爆上下文）
memory = ConversationBufferMemory()

# 2. 窗口记忆（只保留最近 5 轮，省 Token）
memory = ConversationBufferWindowMemory(k=5)

# 3. Token 限制记忆（按 Token 数截断）
memory = ConversationTokenBufferMemory(
    llm=ChatOpenAI(model="gpt-4"),
    max_token_limit=1000
)

# 4. 摘要记忆（把历史对话压缩成摘要）
memory = ConversationSummaryMemory(
    llm=ChatOpenAI(model="gpt-4")
)

conversation = ConversationChain(
    llm=ChatOpenAI(model="gpt-4", temperature=0),
    memory=memory,
    verbose=True,
)

# 多轮对话，记忆自动维护
conversation.predict(input="我叫小明，今年 25 岁")
conversation.predict(input="我喜欢 Python 编程")
print(conversation.predict(input="我叫什么？多大了？"))  # 能回忆起前面说的

from langchain_core.tools import tool

@tool
def get_weather(city: str) -> str:
    """获取指定城市的天气信息。输入城市名称，返回天气描述。"""
    # 实际项目中这里调用真实天气 API
    weather_map = {"北京": "晴 25°C", "上海": "多云 28°C", "广州": "雷雨 30°C"}
    return weather_map.get(city, f"{city}：暂无天气数据")

@tool
def calculate(expression: str) -> str:
    """计算数学表达式的结果。输入如 '2+3*4'，返回计算结果。"""
    try:
        result = eval(expression)  # 生产环境请用安全的表达式解析器
        return f"{expression} = {result}"
    except Exception as e:
        return f"计算失败：{e}"

@tool
def search_wiki(query: str) -> str:
    """搜索维基百科获取知识。输入搜索关键词。"""
    return f"关于'{query}'的维基百科摘要：..."

tools = [get_weather, calculate, search_wiki]

from langchain.agents import create_agent

agent = create_agent(
    model="openai:gpt-5.5",
    tools=[get_weather, calculate, search_wiki],
    system_prompt="你是一个能查天气、算数学、搜百科的助手，回答要简洁",
)

result = agent.invoke(
    {"messages": [{"role": "user", "content": "北京天气怎么样？2的10次方是多少？"}]}
)
# Agent 会自主决定调用哪些工具、按什么顺序调用
print(result["messages"][-1].content)

from langchain_openai import ChatOpenAI

llm = ChatOpenAI(model="gpt-4")
# bind_tools 把工具信息注入模型，模型能"看到"工具并决定调用
llm_with_tools = llm.bind_tools([get_weather, calculate])

response = llm_with_tools.invoke("北京天气怎么样？")
if response.tool_calls:
    for tc in response.tool_calls:
        print(f"模型决定调用: {tc['name']}, 参数: {tc['args']}")

from langchain_core.callbacks import BaseCallbackHandler

class MyCallbackHandler(BaseCallbackHandler):
    def on_llm_start(self, serialized, prompts, **kwargs):
        print(f"[LLM 开始] 模型: {serialized.get('name')}")

    def on_llm_new_token(self, token, **kwargs):
        # 流式输出时，每个 token 都会触发
        print(token, end="", flush=True)

    def on_llm_end(self, response, **kwargs):
        print(f"\n[LLM 结束] 耗时统计完成")

    def on_tool_start(self, serialized, input_str, **kwargs):
        print(f"[工具开始] {serialized['name']}: {input_str}")

    def on_tool_end(self, output, **kwargs):
        print(f"[工具结束] 返回: {output}")

    def on_chain_error(self, error, **kwargs):
        print(f"[链错误] {error}")

# 使用回调
llm = ChatOpenAI(model="gpt-4", streaming=True,
                 callbacks=[MyCallbackHandler()])
llm.invoke("用三句话介绍量子计算")

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

# 用管道符 | 串联组件
prompt = ChatPromptTemplate.from_template("请解释什么是 {topic}")
llm = ChatOpenAI(model="gpt-4")
parser = StrOutputParser()

chain = prompt | llm | parser

# 统一接口调用
result = chain.invoke({"topic": "LCEL"})
print(result)

# 流式调用（Token 级输出）
for chunk in chain.stream({"topic": "LCEL"}):
    print(chunk, end="", flush=True)

# 批量调用
results = chain.batch([{"topic": "RAG"}, {"topic": "Agent"}, {"topic": "Memory"}])
for r in results:
    print(r[:50], "...")

from langchain_core.runnables import RunnableParallel, RunnablePassthrough

summary_prompt = ChatPromptTemplate.from_template("总结：{input}")
keyword_prompt = ChatPromptTemplate.from_template("提取3个关键词：{input}")
sentiment_prompt = ChatPromptTemplate.from_template("判断情感(正面/负面/中性)：{input}")

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

# 三条链并行执行，共享同一个输入
parallel_chain = RunnableParallel(
    summary=summary_prompt | llm | parser,
    keywords=keyword_prompt | llm | parser,
    sentiment=sentiment_prompt | llm | parser,
    # RunnablePassthrough 透传原始输入
    original=RunnablePassthrough(),
)

result = parallel_chain.invoke({
    "input": "LangChain 是一个强大的 LLM 应用开发框架，社区活跃但学习曲线较陡。"
})
print(result["summary"])
print(result["keywords"])
print(result["sentiment"])

# 经典 RAG 模式：检索结果 + 原始问题一起喂给模型
rag_chain = (
    RunnablePassthrough.assign(context=retriever | format_docs)
    | prompt
    | llm
    | parser
)
# RunnablePassthrough.assign() 在不丢失原始输入的基础上，追加 context 字段

from langchain_core.runnables import RunnableWithFallbacks

# 主模型失败时，自动切到备用模型
primary_llm = ChatOpenAI(model="gpt-4")       # 可能限流
fallback_llm = ChatOpenAI(model="gpt-3.5-turbo")  # 更便宜更稳定

llm_with_fallback = primary_llm.with_fallbacks([fallback_llm])

chain = prompt | llm_with_fallback | parser
# 如果 gpt-4 限流报错，自动用 gpt-3.5-turbo 重试

from langchain_core.runnables import RunnableLambda

# 任何函数都能用 RunnableLambda 包装成 Runnable
def extract_user_intent(text: str) -> dict:
    """简单意图识别"""
    if "?" in text or "？" in text:
        return {"intent": "question", "text": text}
    return {"intent": "statement", "text": text}

intent_runnable = RunnableLambda(extract_user_intent)

# 然后就能用管道符串
chain = intent_runnable | prompt | llm | parser

# === 传统 Chains 写法 ===
from langchain.chains import LLMChain
chain_old = LLMChain(
    llm=ChatOpenAI(model="gpt-4"),
    prompt=ChatPromptTemplate.from_template("翻译成英文：{text}"),
    verbose=True,
)
result = chain_old.run(text="你好世界")

# === LCEL 写法 ===
chain_new = (
    ChatPromptTemplate.from_template("翻译成英文：{text}")
    | ChatOpenAI(model="gpt-4")
    | StrOutputParser()
)
result = chain_new.invoke({"text": "你好世界"})
# 更短、更声明式、还免费获得了 stream/batch/async 能力

from typing import TypedDict, Annotated
import operator

class AgentState(TypedDict):
    messages: Annotated[list, operator.add]  # 累加模式：新消息追加到列表
    current_step: str                         # 覆盖模式：新值替换旧值
    results: Annotated[list, operator.add]    # 累加模式

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

llm = ChatOpenAI(model="gpt-4")

def analyze_node(state: AgentState):
    """分析用户输入，确定意图"""
    user_msg = state["messages"][-1].content
    intent = "search" if "查" in user_msg else "chat"
    return {"current_step": intent}  # 只返回要更新的字段

def llm_node(state: AgentState):
    """使用 LLM 生成回复"""
    response = llm.invoke(state["messages"])
    return {"messages": [response]}  # Reducer 会自动累加到 messages

def search_node(state: AgentState):
    """执行检索"""
    query = state["messages"][-1].content
    docs = retriever.invoke(query)
    context = "\n".join(d.page_content[:200] for d in docs)
    response = llm.invoke([
        HumanMessage(content=f"根据以下资料回答：{context}\n问题：{query}")
    ])
    return {"messages": [response], "results": [context]}

from langgraph.graph import StateGraph, START, END

def route_by_intent(state: AgentState) -> str:
    """根据意图路由到不同节点"""
    if state["current_step"] == "search":
        return "search_node"
    else:
        return "llm_node"

graph = StateGraph(AgentState)
graph.add_node("analyze", analyze_node)
graph.add_node("search_node", search_node)
graph.add_node("llm_node", llm_node)

graph.add_edge(START, "analyze")
graph.add_conditional_edges("analyze", route_by_intent, {
    "search_node": "search_node",
    "llm_node": "llm_node",
})
graph.add_edge("search_node", END)
graph.add_edge("llm_node", END)

# 基础编译
app = graph.compile()

# 带 Checkpointer 编译（启用持久化）
from langgraph.checkpoint.memory import InMemorySaver
app = graph.compile(checkpointer=InMemorySaver())

# 带断点编译（在指定节点前暂停）
app = graph.compile(checkpointer=checkpointer, interrupt_before=["approval_node"])

config = {"configurable": {"thread_id": "user-123"}}

result = app.invoke(
    {"messages": [HumanMessage(content="帮我查一下 LangChain 是什么")]},
    config=config,
)
print(result["messages"][-1].content)

from langgraph.checkpoint.postgres import PostgresSaver
from langgraph.store.memory import InMemoryStore

# 生产级持久化
checkpointer = PostgresSaver.from_conn_string("postgresql://...")

# 长期记忆存储（跨对话线程）
store = InMemoryStore()  # 生产用 PostgresStore
# 记住用户偏好
store.put(("user", "123"), "preferences", {"language": "zh", "tone": "formal"})

app = graph.compile(checkpointer=checkpointer, store=store)

from langgraph.types import interrupt, Command

def approval_node(state: AgentState):
    # 暂停执行，等待人工审核
    approved = interrupt("请确认是否执行此操作？")
    if approved:
        return {"approved": True, "result": "操作已执行"}
    else:
        return {"approved": False, "result": "操作已取消"}

# 第一次调用：遇到 interrupt 暂停
config = {"configurable": {"thread_id": "approval-1"}}
result = app.invoke({"messages": [HumanMessage("删除数据库")]}, config=config)
# 此时执行暂停在 approval_node，等待人工输入

# 恢复执行：传入人工审核结果
result = app.invoke(Command(resume=True), config=config)

# Token 级流式（用户看到打字机效果）
for event in app.stream(
    {"messages": [HumanMessage("讲个故事")]},
    config=config,
    stream_mode="messages",
):
    # event 是 (message, metadata) 元组
    msg, meta = event
    if msg.content:
        print(msg.content, end="", flush=True)

# 状态增量流式（调试用）
for event in app.stream(
    {"messages": [HumanMessage("查 LangChain")]},
    config=config,
    stream_mode="updates",
):
    for node_name, update in event.items():
        print(f"[节点 {node_name}] 更新: {list(update.keys())}")

def supervisor(state: AgentState):
    """主管 Agent：决定将任务分配给哪个专家"""
    response = llm.invoke([
        SystemMessage(content="""你是一个任务调度器。
        可选专家：researcher(查资料)、writer(写文章)、reviewer(审稿)。
        只返回专家名字。"""),
        *state["messages"]
    ])
    return {"next_agent": response.content.strip().lower()}

def researcher(state: AgentState):
    """研究员 Agent"""
    docs = retriever.invoke(state["messages"][-1].content)
    return {"messages": [AIMessage(content=f"研究完成：{docs[0].page_content[:200]}")]}

def writer(state: AgentState):
    """写作 Agent"""
    response = llm.invoke(state["messages"])
    return {"messages": [response]}

def reviewer(state: AgentState):
    """审稿 Agent"""
    response = llm.invoke([
        SystemMessage(content="你是审稿人，给出修改建议"),
        *state["messages"]
    ])
    return {"messages": [response]}

# 构建图
graph = StateGraph(AgentState)
graph.add_node("supervisor", supervisor)
graph.add_node("researcher", researcher)
graph.add_node("writer", writer)
graph.add_node("reviewer", reviewer)

graph.add_edge(START, "supervisor")
graph.add_conditional_edges("supervisor", lambda s: s["next_agent"], {
    "researcher": "researcher",
    "writer": "writer",
    "reviewer": "reviewer",
    "FINISH": END,
})
# 每个专家做完回到主管
graph.add_edge("researcher", "supervisor")
graph.add_edge("writer", "supervisor")
graph.add_edge("reviewer", "supervisor")

app = graph.compile()

graph.add_edge(START, "researcher_a")  # 并行执行
graph.add_edge(START, "researcher_b")  # 并行执行
graph.add_edge("researcher_a", "synthesizer")
graph.add_edge("researcher_b", "synthesizer")
graph.add_edge("synthesizer", END)

def should_continue(state: AgentState) -> str:
    if state.get("iteration_count", 0) >= 3:
        return "finalize"
    if state.get("critique_approved", False):
        return "finalize"
    return "revise"

graph.add_edge(START, "generate")
graph.add_edge("generate", "critique")
graph.add_conditional_edges("critique", should_continue, {
    "revise": "revise",
    "finalize": "finalize"
})
graph.add_edge("revise", "critique")  # 循环
graph.add_edge("finalize", END)

# 用 YAML DSL 定义工作流，动态解析为 LangGraph 图
import yaml

workflow_yaml = """
nodes:
  - id: start
    type: input
  - id: retrieve
    type: knowledge-retrieval
    config:
      vectorstore: chroma
      k: 3
  - id: generate
    type: llm
    config:
      model: gpt-4
      prompt_template: "根据资料回答：{context}\\n问题：{question}"
  - id: review
    type: human-review
  - id: answer
    type: output

edges:
  - {from: start, to: retrieve}
  - {from: retrieve, to: generate}
  - {from: generate, to: review}
  - {from: review, to: answer}
"""

# 动态构建图（伪代码，展示思路）
def build_graph_from_dsl(dsl):
    spec = yaml.safe_load(dsl)
    graph = StateGraph(AgentState)
    for node in spec["nodes"]:
        graph.add_node(node["id"], create_node_function(node))
    for edge in spec["edges"]:
        graph.add_edge(edge["from"], edge["to"])
    return graph.compile(checkpointer=PostgresSaver(...))

app = build_graph_from_dsl(workflow_yaml)

import os

# 设置追踪环境变量（只需设置一次，全局生效）
os.environ["LANGSMITH_TRACING"] = "true"
os.environ["LANGSMITH_API_KEY"] = "your-langsmith-key"
os.environ["LANGSMITH_PROJECT"] = "langchain-project"

# 设置后，所有 LangChain/LangGraph 的调用都会自动被追踪
# 不需要改任何业务代码
from langchain_openai import ChatOpenAI
llm = ChatOpenAI(model="gpt-4")
response = llm.invoke("什么是 RAG？")
# ↑ 这次调用会自动出现在 LangSmith 平台，含完整输入输出和耗时

# 定义 Agent
from langchain.agents import create_agent
from langchain_core.tools import tool

@tool
def search_db(query: str) -> str:
    """搜索数据库"""
    return f"查询结果：{query} 的数据"

agent = create_agent(
    model="openai:gpt-4",
    tools=[search_db],
    system_prompt="你是一个数据查询助手",
)

# 执行（自动追踪）
result = agent.invoke({
    "messages": [{"role": "user", "content": "查一下上月销售额"}]
})

# 在 LangSmith 平台你会看到这样的追踪树：
#
#   Agent Run (total: 3.2s)
#   ├── Step 1: LLM 决策 (1.1s)
#   │   └── 输出：调用 search_db("上月销售额")
#   ├── Step 2: 工具执行 search_db (0.3s)
#   │   └── 返回："查询结果：上月销售额的数据"
#   ├── Step 3: LLM 总结 (1.5s)
#   │   └── 输出："上月销售额为..."
#   └── Step 4: 返回最终回复 (0.3s)

from langsmith import Client
from langsmith.evaluation import evaluate

client = Client()

# 创建数据集
dataset = client.create_dataset("rag-eval-v1")

# 添加测试用例
client.create_examples(
    inputs=[
        {"question": "LangChain 的核心组件有哪些？"},
        {"question": "什么是 LCEL？"},
        {"question": "LangGraph 和 LangChain 的区别？"},
    ],
    outputs=[
        {"answer": "Model I/O, Data Connection, Chains, Memory, Agents, Callbacks"},
        {"answer": "LangChain Expression Language，用管道符组合组件"},
        {"answer": "LangChain 提供积木，LangGraph 提供图编排脚手架"},
    ],
    dataset_id=dataset.id,
)

# 评估你的 RAG 链
def accuracy_evaluator(run, example):
    """自定义评估器：检查答案是否包含关键信息"""
    prediction = run.outputs.get("output", "")
    expected = example.outputs.get("answer", "")
    score = 1.0 if any(kw in prediction for kw in expected.split()) else 0.0
    return {"key": "accuracy", "score": score}

results = evaluate(
    lambda inputs: rag_chain.invoke(inputs["question"]),
    data="rag-eval-v1",
    evaluators=[accuracy_evaluator],
)
print(f"准确率: {results['aggregate_metrics']['accuracy']:.1%}")

# langgraph.json - 部署配置文件
{
  "dependencies": ["."],
  "graphs": {
    "my_agent": "./app/agent.py:graph"
  },
  "env": ".env"
}

# app/agent.py
from langgraph.graph import StateGraph, START, END
from langgraph.checkpoint.postgres import PostgresSaver

# ... 定义 State、Nodes、Edges（同第五章）...

graph = StateGraph(AgentState)
# ... add_node / add_edge ...

# 编译成可部署的图
checkpointer = PostgresSaver.from_conn_string(os.environ["DATABASE_URL"])
app = graph.compile(checkpointer=checkpointer)

# 安装 LangGraph CLI
pip install langgraph-cli

# 本地开发运行
langgraph dev

# 构建部署镜像
langgraph build -t my-agent:latest

# 部署到 LangGraph Cloud
langgraph deploy

import requests

# 同步调用
response = requests.post(
    "http://localhost:8000/threads/my-agent/runs",
    json={
        "assistant_id": "my_agent",
        "input": {"messages": [{"role": "user", "content": "你好"}]},
        "config": {"configurable": {"thread_id": "session-1"}},
    }
)
print(response.json())

# 流式调用
import sseclient
response = requests.post(
    "http://localhost:8000/threads/my-agent/runs/stream",
    json={
        "assistant_id": "my_agent",
        "input": {"messages": [{"role": "user", "content": "讲个故事"}]},
        "stream_mode": "messages",
    },
    stream=True,
)
client = sseclient.SSEClient(response)
for event in client.events():
    print(event.data, end="", flush=True)

# 手搓一个最小 ReAct Agent（不用任何框架）
import openai

def react_agent(question, tools, max_iter=5):
    """手搓 ReAct：思考→行动→观察→思考..."""
    messages = [
        {"role": "system", "content": f"""你是一个能使用工具的助手。
可用工具：{tools}
当你要使用工具时，输出 JSON：{{"action": "工具名", "input": "参数"}}
当你有最终答案时，输出：{{"action": "final_answer", "input": "答案"}}"""},
        {"role": "user", "content": question},
    ]
    for i in range(max_iter):
        resp = openai.chat.completions.create(model="gpt-4", messages=messages)
        thought = resp.choices[0].message.content
        messages.append({"role": "assistant", "content": thought})
        # 解析 action 并执行（省略解析逻辑）
        if "final_answer" in thought:
            return thought
        # 执行工具，把结果作为 observation 喂回去
        messages.append({"role": "user", "content": f"观察结果：{tool_result}"})
    return "达到最大迭代次数"

from typing import TypedDict, Annotated
import operator

# 多 Schema 设计：分离输入、输出、内部状态
class InputState(TypedDict):
    """用户输入的 State（只含用户能提供的字段）"""
    question: str

class OutputState(TypedDict):
    """输出给用户的 State（只含需要返回的字段）"""
    answer: str
    sources: list[str]

class OverallState(InputState, OutputState):
    """内部完整 State（含所有中间字段）"""
    messages: Annotated[list, operator.add]
    retrieved_docs: Annotated[list, operator.add]
    iteration_count: int  # 覆盖模式

# 自定义 Reducer：解决并行节点覆盖问题
def merge_dicts(left, right):
    """并行节点的 dict 合并：递归合并而非覆盖"""
    if left is None:
        return right
    if right is None:
        return left
    result = {**left}
    for k, v in right.items():
        if k in result and isinstance(result[k], list):
            result[k] = result[k] + v
        else:
            result[k] = v
    return result

class ParallelState(TypedDict):
    # 多个并行节点都写这个字段，用自定义 Reducer 合并
    results: Annotated[dict, merge_dicts]

from langgraph.checkpoint.postgres import PostgresSaver
from langgraph.store.postgres import PostgresStore
from contextlib import asynccontextmanager

# 生产级配置
DB_URI = "postgresql://user:pass@localhost:5432/langgraph"

# 持久化 + 长期记忆
checkpointer = PostgresSaver.from_conn_string(DB_URI)
store = PostgresStore.from_conn_string(DB_URI)

app = graph.compile(checkpointer=checkpointer, store=store)

# 长期记忆：记住用户偏好，跨对话线程
def save_user_preference(user_id, key, value):
    store.put(("user", user_id), key, {"value": value})

def get_user_preference(user_id, key):
    item = store.get(("user", user_id), key)
    return item.value.get("value") if item else None

from langgraph.config import get_stream_writer

def long_running_node(state):
    """模拟长时间运行节点，发送自定义进度"""
    writer = get_stream_writer()
    
    steps = ["加载模型", "检索文档", "生成回复", "格式化输出"]
    for i, step in enumerate(steps):
        writer({"progress": f"步骤 {i+1}/{len(steps)}: {step}"})
        # ... 实际工作 ...
    
    return {"answer": "完成"}

# 客户端接收自定义流
for mode, chunk in app.stream(
    {"question": "复杂问题"},
    config=config,
    stream_mode=["messages", "custom"],
):
    if mode == "custom":
        print(f"[进度] {chunk.get('progress')}")
    elif mode == "messages":
        msg = chunk[0]
        if msg.content:
            print(msg.content, end="", flush=True)

from langgraph.types import interrupt, Command

def dangerous_action_node(state):
    """执行危险操作前的审核"""
    action_desc = f"即将执行：{state['pending_action']}"
    
    # interrupt 暂停，把决策权交给人类
    approval = interrupt({
        "type": "action_approval",
        "description": action_desc,
        "severity": "high",
    })
    
    if approval == "approved":
        # 执行危险操作
        result = execute_dangerous_action(state)
        return {"result": result, "status": "executed"}
    elif approval == "modified":
        # 人类修改了操作
        result = execute_dangerous_action(state, modified=True)
        return {"result": result, "status": "executed_modified"}
    else:
        return {"result": "操作被拒绝", "status": "rejected"}

# 恢复执行
result = app.invoke(Command(resume="approved"), config=config)

import os
os.environ["LANGSMITH_TRACING"] = "true"
os.environ["LANGSMITH_API_KEY"] = "ls__xxx"
os.environ["LANGSMITH_PROJECT"] = "prod-agent-v2"

# 给重要调用打标签，方便在 LangSmith 筛选
from langchain_core.tracers.context import tracing_v2_enabled

with tracing_v2_enabled(
    project_name="prod-agent-v2",
    tags=["customer-service", "v2"],
    metadata={"user_id": "12345", "session": "abc"},
):
    result = agent.invoke({"messages": [...]})
    # 这次调用在 LangSmith 会带上 tag 和 metadata，方便筛选分析

from langchain_core.tools import tool, BaseToolkit

# 工具描述黄金法则：写"什么场景该调我"，不写"我是干嘛的"
@tool
def query_sales_db(date_range: str, region: str) -> str:
    """查询销售数据库。
    
    什么时候用我：当用户问"某段时间某个地区的销售额/订单量/客户数"时。
    参数说明：
    - date_range: 日期范围，格式 "2024-01-01~2024-01-31" 或 "最近7天"
    - region: 地区，如 "华东"、"华南"、"全国"
    
    返回：包含销售额、订单量、客单价的 JSON。
    """
    return query_db(date_range, region)

# 工具不宜过多：超过 10 个工具时，模型选择准确率显著下降
# 解决方案：用路由 Agent 先选工具集，再传给执行 Agent

# 子图：把复杂工作流拆成可复用的模块
def build_research_subgraph():
    """研究子图：检索→分析→总结"""
    sub_graph = StateGraph(ResearchState)
    sub_graph.add_node("retrieve", retrieve_node)
    sub_graph.add_node("analyze", analyze_node)
    sub_graph.add_node("summarize", summarize_node)
    sub_graph.add_edge(START, "retrieve")
    sub_graph.add_edge("retrieve", "analyze")
    sub_graph.add_edge("analyze", "summarize")
    sub_graph.add_edge("summarize", END)
    return sub_graph.compile()

# 主图中嵌入子图
main_graph = StateGraph(OverallState)
main_graph.add_node("research", build_research_subgraph())  # 子图作为节点
main_graph.add_node("write", write_node)
main_graph.add_edge(START, "research")
main_graph.add_edge("research", "write")
main_graph.add_edge("write", END)

import asyncio
from tenacity import retry, stop_after_attempt, wait_exponential

# 异步 + 重试
@retry(stop=stop_after_attempt(3), wait=wait_exponential(min=1, max=10))
async def call_agent_async(user_input, thread_id):
    config = {"configurable": {"thread_id": thread_id}}
    result = await app.ainvoke(
        {"messages": [HumanMessage(content=user_input)]},
        config=config,
    )
    return result["messages"][-1].content

# 并发处理多个请求
async def handle_batch(inputs):
    tasks = [call_agent_async(inp["text"], inp["thread_id"]) for inp in inputs]
    return await asyncio.gather(*tasks, return_exceptions=True)

# 防止 Agent 无限循环：加迭代计数器
def should_continue(state: AgentState) -> str:
    count = state.get("iteration_count", 0)
    if count >= 10:  # 最大 10 轮
        return "END"
    return "continue"

# 上下文压缩：消息太多时自动摘要
def compress_if_needed(state: AgentState):
    messages = state["messages"]
    if len(messages) > 20:  # 超过 20 条消息就压缩
        summary = llm.invoke([
            SystemMessage("把以下对话压缩成一段摘要"),
            *messages[:-4],  # 保留最近 4 条不压缩
        ])
        state["messages"] = [summary] + messages[-4:]
    return state