Part 9 · LangChain 框架

LangChain = 一个"乐高框架"，把 LLM 应用常用的零件做成了标准积木：  Prompt模板 / 模型封装 / 记忆 / 检索 / 工具 / 链LangGraph = LangChain 家的"图编排"，做 Agent 状态机（站8用）

pip install langchain langchain-openai langchain-community

import osfrom dotenv import load_dotenvfrom langchain_openai import ChatOpenAIload_dotenv()# 用 LangChain 包一层 DeepSeekllm = ChatOpenAI(    model="deepseek-chat",    api_key=os.getenv("DEEPSEEK_API_KEY"),    base_url="https://api.deepseek.com",)resp = llm.invoke("你好，用一句话介绍你自己")print(resp.content)

LLM          老的纯文本接口，输入输出都是字符串ChatModel    现在的对话接口，输入是消息列表（system/user/assistant），更强大

import osfrom dotenv import load_dotenvfrom langchain_openai import ChatOpenAIload_dotenv()llm = ChatOpenAI(model="deepseek-chat", api_key=os.getenv("DEEPSEEK_API_KEY"), base_url="https://api.deepseek.com")messages = [    {"role": "system", "content": "你是资深中文编辑，只改错别字和病句，不改变风格。"},    {"role": "user", "content": "我今天去了公司，发现会议室以经被别人占用了。"},]print(llm.invoke(messages).content)

import osfrom dotenv import load_dotenvfrom langchain_openai import ChatOpenAIfrom langchain_core.prompts import ChatPromptTemplateload_dotenv()llm = ChatOpenAI(model="deepseek-chat", api_key=os.getenv("DEEPSEEK_API_KEY"), base_url="https://api.deepseek.com")# 定义模板：变量用 {xxx} 占位prompt = ChatPromptTemplate.from_messages([    ("system", "你是{topic}专家，回答要专业且简短。"),    ("user", "请解释：{question}"),])# 模板 + 模型 = 一条链chain = prompt | llm# 不同参数复用同一条链for t, q in [("健身", "深蹲怎么练"), ("编程", "什么是递归")]:    print(t, "=>", chain.invoke({"topic": t, "question": q}).content, "\n")

import osfrom dotenv import load_dotenvfrom langchain_openai import ChatOpenAIfrom langchain_core.prompts import ChatPromptTemplatefrom langchain_core.output_parsers import JsonOutputParserload_dotenv()llm = ChatOpenAI(model="deepseek-chat", api_key=os.getenv("DEEPSEEK_API_KEY"), base_url="https://api.deepseek.com")parser = JsonOutputParser()  # 自动把输出解析成 dictprompt = ChatPromptTemplate.from_messages([    ("system", "提取信息，只输出 JSON，不要其他文字。字段：name, price, category。"),    ("user", "{text}"),])chain = prompt | llm | parserresult = chain.invoke({"text": "这款机械键盘卖399元，属于外设类"})print(result)print("价格:", result["price"])

import osfrom dotenv import load_dotenvfrom langchain_openai import ChatOpenAIfrom langchain_core.messages import HumanMessage, AIMessageload_dotenv()llm = ChatOpenAI(model="deepseek-chat", api_key=os.getenv("DEEPSEEK_API_KEY"), base_url="https://api.deepseek.com")# 手动维护对话历史：这就是"记忆"的最原始形态history = [    HumanMessage("我叫小明"),    AIMessage("你好小明，很高兴认识你！"),]history.append(HumanMessage("我叫什么名字？"))print(llm.invoke(history).content)   # 应该记得"小明"# 清空历史再问print(llm.invoke([HumanMessage("我叫什么名字？")]).content)  # 不知道了

LCEL（LangChain Expression Language）prompt | llm | parser= 数据像水流过管道：prompt 生成消息 → llm 生成回答 → parser 解析好处：好读、好组合、自带流式和重试

import osfrom dotenv import load_dotenvfrom langchain_openai import ChatOpenAIfrom langchain_core.prompts import ChatPromptTemplateload_dotenv()llm = ChatOpenAI(model="deepseek-chat", api_key=os.getenv("DEEPSEEK_API_KEY"), base_url="https://api.deepseek.com")prompt = ChatPromptTemplate.from_messages([    ("system", "把用户输入翻译成英文，只输出翻译结果。"),    ("user", "{text}"),])chain = prompt | llm# 流式输出（LCEL 自带）for chunk in chain.stream({"text": "你好，很高兴认识你"}):    print(chunk.content, end="", flush=True)

import osfrom dotenv import load_dotenvfrom langchain_openai import ChatOpenAI, OpenAIEmbeddingsfrom langchain_community.vectorstores import Chromafrom langchain_core.prompts import ChatPromptTemplatefrom langchain_text_splitters import RecursiveCharacterTextSplitterfrom langchain_community.document_loaders import TextLoaderload_dotenv()llm = ChatOpenAI(model="deepseek-chat", api_key=os.getenv("DEEPSEEK_API_KEY"), base_url="https://api.deepseek.com")# 用 DeepSeek 的 embedding 接口（或换 bge）。这里用 OpenAI 兼容方式：from langchain_openai import OpenAIEmbeddingsembeddings = OpenAIEmbeddings(    model="text-embedding-v1",   # 用 DashScope 的兼容接口时需要；简单起见先注释    api_key=os.getenv("DEEPSEEK_API_KEY"),    base_url="https://api.deepseek.com",)

# 如果上面 DashScope 不顺，就用Part 5 的 rag_tool 继续，LangChain 只学"串链"这部分：prompt = ChatPromptTemplate.from_messages([    ("system", "根据资料回答，资料没有就说没有：\n{context}"),    ("user", "{question}"),])chain = prompt | llm# context 从站5的检索结果来

import osfrom dotenv import load_dotenvfrom langchain_openai import ChatOpenAIfrom langchain_core.tools import toolload_dotenv()llm = ChatOpenAI(model="deepseek-chat", api_key=os.getenv("DEEPSEEK_API_KEY"), base_url="https://api.deepseek.com")@tooldef add(a: int, b: int) -> int:    """两数相加。用这个工具做精确计算，避免模型算错。"""    return a + bprint("工具名:", add.name)print("调用:", add.invoke({"a": 3, "b": 4}))

import osfrom dotenv import load_dotenvfrom langchain_openai import ChatOpenAIfrom langchain_core.tools import toolfrom langchain.agents import create_agentload_dotenv()llm = ChatOpenAI(model="deepseek-chat", api_key=os.getenv("DEEPSEEK_API_KEY"), base_url="https://api.deepseek.com")@tooldef multiply(a: int, b: int) -> int:    """两数相乘。"""    return a * btools = [multiply]# LangChain 1.x 用 create_agent，不再需要 create_tool_calling_agent + AgentExecutor# system_prompt 替代旧版的 ChatPromptTemplate# debug=True 打印工具调用过程（替代旧版 verbose=True）agent = create_agent(    model=llm,    tools=tools,    system_prompt="你是计算助手，需要时调用工具。",    debug=True,)result = agent.invoke({"messages": [{"role": "user", "content": "123 乘以 456 等于多少？"}]})print(result["messages"][-1].content)

from typing import TypedDictfrom langgraph.graph import StateGraph, ENDclass State(TypedDict):    total: intdef add_one(state: State):    return {"total": state["total"] + 1}# 建图：两个节点顺序执行g = StateGraph(State)g.add_node("a", add_one)g.add_node("b", add_one)g.set_entry_point("a")g.add_edge("a", "b")g.add_edge("b", END)app = g.compile()print(app.invoke({"total": 0}))   # 预期 {'total': 2}

TAVILY_API_KEY=tvly-你的key

import osfrom dotenv import load_dotenvfrom langchain_community.tools import TavilySearchResultsload_dotenv()tool = TavilySearchResults(max_results=3)result = tool.invoke("2026年大模型行业有什么大新闻")for r in result:    print(r["title"], "|", r["url"])

import osfrom dotenv import load_dotenvfrom langchain_openai import ChatOpenAIfrom langchain_community.document_loaders import TextLoaderfrom langchain_text_splitters import RecursiveCharacterTextSplitterfrom langchain_community.vectorstores import Chromafrom langchain_community.embeddings import HuggingFaceEmbeddingsfrom langchain_core.prompts import ChatPromptTemplateload_dotenv()llm = ChatOpenAI(model="deepseek-chat", api_key=os.getenv("DEEPSEEK_API_KEY"), base_url="https://api.deepseek.com")# 1. 加载 + 切分loader = TextLoader("rag_data/company.md", encoding="utf-8")docs = loader.load()splitter = RecursiveCharacterTextSplitter(chunk_size=200, chunk_overlap=40)chunks = splitter.split_documents(docs)# 2. 向量化 + 入库（用 bge，免 DashScope Key）embeddings = HuggingFaceEmbeddings(model_name="BAAI/bge-small-zh-v1.5")db = Chroma.from_documents(chunks, embeddings, persist_directory="lc_db")# 3. 检索 + 生成retriever = db.as_retriever(search_kwargs={"k": 3})prompt = ChatPromptTemplate.from_messages([    ("system", "根据资料回答，没有就说没有：\n{context}"),    ("human", "{question}"),])def ask(q):    context = "\n".join(d.page_content for d in retriever.invoke(q))    return llm.invoke(prompt.format_messages(question=q, context=context)).contentprint(ask("迟到会怎么样？"))

import osfrom dotenv import load_dotenvfrom langchain_openai import ChatOpenAIfrom langchain_community.vectorstores import Chromafrom langchain_community.embeddings import HuggingFaceEmbeddingsfrom langchain_core.messages import HumanMessage, AIMessageload_dotenv()llm = ChatOpenAI(model="deepseek-chat", api_key=os.getenv("DEEPSEEK_API_KEY"), base_url="https://api.deepseek.com")embeddings = HuggingFaceEmbeddings(model_name="BAAI/bge-small-zh-v1.5")db = Chroma(persist_directory="lc_db", embedding_function=embeddings)retriever = db.as_retriever(search_kwargs={"k": 3})history = []  # 记忆def chat(user_input):    # 检索    context = "\n".join(d.page_content for d in retriever.invoke(user_input))    # 拼历史 + 资料 + 问题    messages = [{"role": "system", "content": f"根据资料回答：\n{context}"}]    messages += history    messages.append({"role": "user", "content": user_input})    resp = llm.invoke(messages).content    history.append(HumanMessage(user_input))    history.append(AIMessage(resp))    return respwhile True:    q = input("你：")    if q == "exit":        break    print("助手：", chat(q))

坑1  DeepSeek 没有 embedding 接口 → 用 bge（HuggingFaceEmbeddings）坑2  Chroma persist 目录重名冲突 → 换目录名或删掉旧的坑3  模型"看不到工具说明" → 工具 docstring 必须写清楚坑4  LCEL 流式报错 → 检查模型是否支持 stream坑5  历史无限增长 → 超长截断/只保留最近 N 轮