LangChain 从入门到实践：由浅入深 + DeepAgent 实战#

# 建议使用虚拟环境
python -m venv .venv && source .venv/bin/activate   # Windows: .venv\Scripts\activate

pip install -U langchain langchain-openai langchain-community langgraph
pip install -U deepagents            # DeepAgent（第 10 章使用）
pip install -U python-dotenv         # 管理密钥
pip install -U faiss-cpu             # 轻量向量库（RAG 用，可选 chroma）

OPENAI_API_KEY=sk-xxxx
# 如使用 LangSmith 可观测：
LANGSMITH_TRACING=true
LANGSMITH_API_KEY=lsv2_xxxx
LANGSMITH_PROJECT=langchain-tutorial

# hello_langchain.py
import os
from dotenv import load_dotenv
from langchain_openai import ChatOpenAI

load_dotenv()

# 1) 创建模型（Model）
llm = ChatOpenAI(model="gpt-4o-mini", temperature=0)

# 2) 调用（自动处理消息格式）
response = llm.invoke("用一句话解释什么是 LangChain？")
print(response.content)

from langchain_core.prompts import ChatPromptTemplate

prompt = ChatPromptTemplate.from_messages([
    ("system", "你是一个乐于助人的翻译员，只输出翻译结果，不要解释。"),
    ("user", "把下面这句话翻译成英文：\n{text}"),
])

# 渲染模板（得到可发给模型的消息列表）
messages = prompt.format_prompt(text="今天天气真好").to_messages()
print(messages)

from langchain_core.output_parsers import StrOutputParser, CommaSeparatedListOutputParser

# 字符串解析（最常用）
str_parser = StrOutputParser()

# 列表解析
list_parser = CommaSeparatedListOutputParser()
print(list_parser.parse("苹果, 香蕉, 橙子"))  # ['苹果', '香蕉', '橙子']

# translator.py
from langchain_core.prompts import ChatPromptTemplate
from langchain_core.output_parsers import StrOutputParser
from langchain_openai import ChatOpenAI

llm = ChatOpenAI(model="gpt-4o-mini")
prompt = ChatPromptTemplate.from_messages([
    ("system", "你是翻译员，只输出译文，不解释。目标语言：{language}"),
    ("user", "{text}"),
])
parser = StrOutputParser()

chain = prompt | llm | parser   # 这就是"链"（下一章详讲）

result = chain.invoke({"language": "English", "text": "LangChain 让构建大模型应用变得简单。"})
print(result)
# -> "LangChain makes building LLM applications simple."

chain = prompt | llm | parser

# review_analyzer.py
from langchain_core.prompts import ChatPromptTemplate
from langchain_openai import ChatOpenAI
from pydantic import BaseModel, Field

# 用 Pydantic 定义结构化输出
class ReviewAnalysis(BaseModel):
    sentiment: str = Field(description="正面 / 负面 / 中性")
    score: float = Field(description="情感评分，0~1，越高越正面")
    keywords: list[str] = Field(description="关键观点词")

llm = ChatOpenAI(model="gpt-4o-mini")
prompt = ChatPromptTemplate.from_messages([
    ("system", "你是电商评论分析专家，严格按 schema 输出 JSON。"),
    ("user", "分析这条评论：\n{review}"),
])

# with_structured_output 让模型直接返回结构化对象
structured_llm = llm.with_structured_output(ReviewAnalysis)
chain = prompt | structured_llm

result = chain.invoke({"review": "快递太慢了，等了一周才到，但产品质量还行。"})
print(result.sentiment, result.score, result.keywords)

# 先翻译，再分析（链中套链）
translate_chain = (
    ChatPromptTemplate.from_template("翻译成英文：{text}")
    | ChatOpenAI(model="gpt-4o-mini")
)

analyze_chain = translate_chain | structured_llm
print(analyze_chain.invoke({"text": "这家店服务态度很差。"}))

# memory_chatbot.py
from langchain_core.prompts import ChatPromptTemplate, MessagesPlaceholder
from langchain_core.chat_history import InMemoryChatMessageHistory
from langchain_core.runnables.history import RunnableWithMessageHistory
from langchain_openai import ChatOpenAI

llm = ChatOpenAI(model="gpt-4o-mini")
prompt = ChatPromptTemplate.from_messages([
    ("system", "你是一个友好的助手。"),
    MessagesPlaceholder("history"),   # 历史消息插入点
    ("user", "{input}"),
])

chain = prompt | llm

# 用 session_id 区分不同用户/会话
store = {}  # 生产环境换成 Redis / 数据库
def get_history(session_id):
    return store.setdefault(session_id, InMemoryChatMessageHistory())

chat = RunnableWithMessageHistory(
    chain,
    get_history,
    input_messages_key="input",
    history_messages_key="history",
)

cfg = {"configurable": {"session_id": "user-1"}}
print(chat.invoke({"input": "我叫小明。"}, config=cfg).content)
print(chat.invoke({"input": "我刚才说我叫什么？"}, config=cfg).content)  # 能记住"小明"

# rag_demo.py
from langchain_community.document_loaders import TextLoader
from langchain.text_splitter import RecursiveCharacterTextSplitter
from langchain_openai import OpenAIEmbeddings, ChatOpenAI
from langchain_community.vectorstores import FAISS
from langchain_core.prompts import ChatPromptTemplate
from langchain_core.runnables import RunnablePassthrough
from langchain_core.output_parsers import StrOutputParser

# 1) 加载
docs = TextLoader("company_faq.txt", encoding="utf-8").load()

# 2) 切分
splits = RecursiveCharacterTextSplitter(chunk_size=500, chunk_overlap=50).split_documents(docs)

# 3) 向量化 + 入库
embeddings = OpenAIEmbeddings()
vectorstore = FAISS.from_documents(splits, embeddings)
retriever = vectorstore.as_retriever(search_kwargs={"k": 3})

# 4) 检索 + 生成
llm = ChatOpenAI(model="gpt-4o-mini")
prompt = ChatPromptTemplate.from_template(
    "根据上下文回答。如果不知道就说不知道。\n上下文：{context}\n问题：{question}"
)

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

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

print(rag_chain.invoke("你们的退款政策是什么？"))

from langchain_core.tools import tool

@tool
def get_weather(city: str) -> str:
    """获取指定城市的天气（示例，返回模拟数据）。"""
    return f"{city}：晴，26°C，微风。"

@tool
def calculator(expression: str) -> str:
    """计算一个数学表达式，如 '23 * 45 + 1'。"""
    return str(eval(expression))

# basic_agent.py
from langchain.agents import create_react_agent, AgentExecutor
from langchain_core.prompts import ChatPromptTemplate
from langchain_openai import ChatOpenAI

llm = ChatOpenAI(model="gpt-4o-mini")
tools = [get_weather, calculator]

prompt = ChatPromptTemplate.from_messages([
    ("system", "你是一个能用工具的助手，思考后再行动。"),
    ("user", "{input}"),
    ("assistant", "{agent_scratchpad}"),  # Agent 的中间思考记录
])

agent = create_react_agent(llm, tools, prompt)
executor = AgentExecutor(agent=agent, tools=tools, verbose=True)

print(executor.invoke({"input": "北京现在多少度？如果温度乘 2 再加 10 是多少？"})["output"])

pip install langchain-mcp-adapters
# stdio 方式拉起本地服务器需要 Node 环境（npx）
# 例如官方文件系统服务器：npx -y @modelcontextprotocol/server-filesystem /tmp/workspace

# mcp_agent.py
import asyncio
from langchain_mcp_adapters.client import MultiServerMCPClient
from langchain.agents import create_agent   # 0.3+ 推荐的统一 Agent 工厂

async def main():
    client = MultiServerMCPClient({
        # ① stdio：本地子进程（如官方 filesystem 服务器）
        "filesystem": {
            "command": "npx",
            "args": ["-y", "@modelcontextprotocol/server-filesystem", "/tmp/workspace"],
            "transport": "stdio",
        },
        # ② http：远程 MCP 服务（Streamable HTTP / SSE）
        "weather": {
            "url": "http://localhost:8000/mcp",
            "transport": "http",
        },
    })

    tools = await client.get_tools()   # 自动拉取所有 MCP 工具并转为 LangChain 工具

    # 与普通 @tool 完全一样喂给 Agent
    agent = create_agent("openai:gpt-4o", tools)
    result = await agent.ainvoke({
        "messages": [{"role": "user", "content": "列出 /tmp/workspace 下的文件，并告诉我天气。"}]
    })
    print(result["messages"][-1].content)
    await client.close()   # 记得关闭，释放子进程

asyncio.run(main())

pip install fastmcp   # FastMCP 3.x

# my_server.py
from fastmcp import FastMCP

mcp = FastMCP("My First MCP Server")

# ① Tool：可被模型调用的"动作"（带类型注解 + docstring，自动生成 schema）
@mcp.tool
def add(a: int, b: int) -> int:
    """把两个整数相加。"""
    return a + b

# ② Resource：只读数据源（URI 模板可取参数，适合暴露静态/背景资料）
@mcp.resource("greeting://{name}")
def greeting(name: str) -> str:
    """按名字返回问候语（只读资源）。"""
    return f"Hello, {name}!"

# ③ Prompt：可复用的提示词模板
@mcp.prompt
def review_code(code: str) -> str:
    """生成一段代码审查提示词。"""
    return f"请审查下面这段代码的性能与安全性：\n{code}"

if __name__ == "__main__":
    mcp.run()   # 默认 stdio 传输

# 方式 A：stdio（本地客户端如 Claude Desktop 默认用这种，进程由客户端拉起）
python my_server.py

# 方式 B：HTTP（远程访问，端点为 http://host:port/mcp）
# 改 my_server.py 末行为：mcp.run(transport="streamable-http", host="0.0.0.0", port=8000)
# 或用 CLI：fastmcp run my_server.py:mcp --transport http --port 8000

# stdio 方式接入（对应 8.3 / 10.8 的 MultiServerMCPClient）
client = MultiServerMCPClient({
    "myserver": {
        "command": "python",
        "args": ["my_server.py"],   # 指向你刚写的文件
        "transport": "stdio",
    }
})
# 或 HTTP 方式：{"myserver": {"url": "http://localhost:8000/mcp", "transport": "http"}}
tools = await client.get_tools()    # 拿到的就是 add / greeting / review_code

# langgraph_demo.py
from typing import TypedDict
from langgraph.graph import StateGraph, END
from langchain_openai import ChatOpenAI

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

class State(TypedDict):
    topic: str
    draft: str

def write(state: State):
    resp = llm.invoke(f"写一段关于 {state['topic']} 的简短介绍")
    return {"draft": resp.content}

def decide(state: State) -> str:
    # 简单规则：含敏感词则退回重写
    return "end" if "机密" not in state["draft"] else "rewrite"

graph = StateGraph(State)
graph.add_node("write", write)
graph.add_conditional_edges("write", decide, {"end": END, "rewrite": "write"})
graph.set_entry_point("write")
app = graph.compile()

print(app.invoke({"topic": "人工智能"}))

pip install -U deepagents langchain-openai
# 可选：代码解释器（v0.6+）
pip install -U "deepagents[quickjs]"   # Python

# deepagent_minimal.py
from deepagents import create_deep_agent

def get_weather(city: str) -> str:
    """获取指定城市的天气。"""
    return f"{city}：晴，26°C。"

agent = create_deep_agent(
    model="openai:gpt-4o",          # 模型字符串格式：厂商:模型名
    tools=[get_weather],
    system_prompt="你是一个有用的助手。",
)

result = agent.invoke({
    "messages": [{"role": "user", "content": "旧金山的天气怎么样？"}]
})
print(result["messages"][-1].content)

# deepagent_research.py
from deepagents import create_deep_agent

agent = create_deep_agent(
    model="openai:gpt-4o",
    system_prompt="你是一个研究助手。请使用文件工具保存中间结果。",
)

# 让 Agent 自己规划、检索、写文件
result = agent.invoke({
    "messages": [{
        "role": "user",
        "content": "调研 'LangGraph 是什么'，把要点写入 research.md，并附上 3 个关键结论。",
    }]
})

# 文件系统后端可用 FilesystemBackend 指定落盘路径（见 10.6）

# deepagent_subagents.py
from deepagents import create_deep_agent

agent = create_deep_agent(
    model="openai:gpt-4o",
    system_prompt="遇到可并行的子问题，用 task 工具委派给子代理。",
)

result = agent.invoke({
    "messages": [{
        "role": "user",
        "content": "分别调研 '向量数据库'、'RAG'、'Agent 评估' 三个主题，各写一段总结，合并到 report.md。",
    }]
})

# deepagent_memory.py
from deepagents import create_deep_agent

agent = create_deep_agent(
    model="openai:gpt-4o",
    system_prompt="记住用户的偏好，并在后续对话中使用。",
)

# 第一次
agent.invoke({"messages": [{"role": "user", "content": "我喜欢用中文，且关注成本优化。"}]})
# 后续新会话（基于 LangGraph store 持久化）
agent.invoke({"messages": [{"role": "user", "content": "帮我选个模型。"}]})  # 会记得你的偏好

# deepagent_hitl.py
from deepagents import create_deep_agent
from langgraph.checkpoint.memory import InMemorySaver

# 让删除/发送等敏感工具在执行前中断审批
agent = create_deep_agent(
    model="openai:gpt-4o",
    system_prompt="对不可逆操作必须先征求用户同意。",
)
checkpointer = InMemorySaver()
app = agent.with_config({"recursion_limit": 50})

# 运行到需要人工审批时，LangGraph 会 interrupt，等待人类 resume

# deepagent_integrate.py
from deepagents import create_deep_agent
from langchain_core.tools import tool
from langchain_openai import ChatOpenAI
from langchain_core.prompts import ChatPromptTemplate

# 复用第 8 章的工具
@tool
def get_weather(city: str) -> str:
    """获取城市天气。"""
    return f"{city}：晴。"

# 复用第 5 章的链，包成工具
translate_chain = (
    ChatPromptTemplate.from_template("翻译成英文：{text}")
    | ChatOpenAI(model="gpt-4o-mini")
)

@tool
def translate(text: str) -> str:
    """把中文翻译成英文。"""
    return translate_chain.invoke({"text": text})

# 全部交给 DeepAgent 统一编排
agent = create_deep_agent(
    model="openai:gpt-4o",
    tools=[get_weather, translate],
    system_prompt="你是一个全能助手，合理使用工具完成任务。",
)

agent.invoke({"messages": [{"role": "user", "content": "把'北京天气不错'翻译成英文，并查北京的天气。"}]})

# deepagent_mcp.py
import asyncio
from langchain_mcp_adapters.client import MultiServerMCPClient
from deepagents import create_deep_agent

async def main():
    # 1) 连接 MCP 服务器（数据库 / API / 文件系统皆可）
    client = MultiServerMCPClient({
        "my_server": {
            "transport": "http",
            "url": "http://localhost:8000/mcp",
        }
    })
    tools = await client.get_tools()   # 转为 LangChain 工具

    # 2) 直接喂给 DeepAgent；MCP 工具 + 内置 harness 工具 + 自定义工具自动融合
    agent = create_deep_agent(
        model="openai:gpt-4o",
        tools=tools,
    )

    # 3) 运行（thread_id 用于记忆隔离，详见 10.5）
    result = await agent.ainvoke(
        {"messages": [{"role": "user", "content": "用 MCP 服务器帮我查一下数据库里最近的 5 笔订单，并整理成表格。"}]},
        config={"configurable": {"thread_id": "1"}},
    )
    print(result["messages"][-1].content)
    await client.close()

asyncio.run(main())

# deepagent_full.py
from deepagents import create_deep_agent
from langchain_core.tools import tool
from langchain_community.vectorstores import FAISS
from langchain_openai import OpenAIEmbeddings

# 1) 准备私有知识库（RAG 检索器）
embeddings = OpenAIEmbeddings()
vectorstore = FAISS.load_local("my_kb", embeddings, allow_dangerous_deserialization=True)
retriever = vectorstore.as_retriever(k=3)

@tool
def search_knowledge_base(query: str) -> str:
    """在公司私有知识库中检索相关信息。"""
    docs = retriever.invoke(query)
    return "\n\n".join(d.page_content for d in docs)

@tool
def save_report(content: str) -> str:
    """把最终报告保存到 report.md。"""
    with open("report.md", "w", encoding="utf-8") as f:
        f.write(content)
    return "已保存 report.md"

# 2) 交给 DeepAgent 统一规划与执行
agent = create_deep_agent(
    model="openai:gpt-4o",
    tools=[search_knowledge_base, save_report],
    system_prompt="你是一个研究助理：先检索内部知识库，再整理成报告并保存。",
)

agent.invoke({
    "messages": [{
        "role": "user",
        "content": "基于内部知识库，整理一份'2026 产品规划要点'报告并保存。",
    }]
})

# 设置环境变量后，所有链/图自动追踪
os.environ["LANGSMITH_TRACING"] = "true"
os.environ["LANGSMITH_API_KEY"] = "lsv2_xxx"
os.environ["LANGSMITH_PROJECT"] = "prod-agents"

# LCEL 链
for chunk in chain.stream({"input": "介绍一下 RAG"}):
    print(chunk, end="", flush=True)

# DeepAgent（v0.6+ 统一事件流）
stream = agent.stream_events(
    {"messages": [{"role": "user", "content": "研究 LangChain 流式"}]},
    version="v3",
)
for msg in stream.messages:
    for delta in msg.text:
        print(delta, end="", flush=True)

# 给模型调用加重试与降级
llm_with_retry = llm.with_retry(retry_if_exception_type=Exception, stop_after_attempt=3)

from langchain_community.document_loaders import DirectoryLoader
from langchain.text_splitter import RecursiveCharacterTextSplitter
from langchain_openai import OpenAIEmbeddings
from langchain_community.vectorstores import FAISS

docs = DirectoryLoader("kb", glob="**/*.md").load()
splits = RecursiveCharacterTextSplitter(chunk_size=800, chunk_overlap=80).split_documents(docs)
FAISS.from_documents(splits, OpenAIEmbeddings()).save_local("my_kb")
print("知识库构建完成")

from deepagents import create_deep_agent
from langchain_core.tools import tool
from langchain_community.vectorstores import FAISS
from langchain_openai import OpenAIEmbeddings

retriever = FAISS.load_local("my_kb", OpenAIEmbeddings(),
                             allow_dangerous_deserialization=True).as_retriever(k=4)

@tool
def search_kb(query: str) -> str:
    """检索公司私有知识库。"""
    return "\n\n".join(d.page_content for d in retriever.invoke(query))

@tool
def save_report(markdown: str) -> str:
    """保存最终报告到 output.md。"""
    with open("output.md", "w", encoding="utf-8") as f:
        f.write(markdown)
    return "报告已保存"

agent = create_deep_agent(
    model="openai:gpt-4o",
    tools=[search_kb, save_report],
    system_prompt="你是资深分析师：检索→分析→生成结构化 Markdown 报告→保存。",
)

agent.invoke({
    "messages": [{
        "role": "user",
        "content": "基于知识库生成《2026 上半年业务复盘与下半年策略》报告，含数据、风险、行动计划三章。",
    }]
})