发布时间： 2025年10月 适用版本： LangChain 1.0.x（Python 3.10+） 前置知识： Python 基础、LLM API 调用基础
2025年10月22日，LangChain 正式发布了 1.0.0 版本——这是该框架自2022年诞生以来的第一个稳定大版本。官方郑重承诺：在发布 2.0 之前，不会有任何破坏性变更（Breaking Changes）。
这是一个重要里程碑，但也带来了困惑：
|
create_react_agent
本文将系统性地回答以上问题，覆盖 LangChain 1.0 的核心概念、API 变化、新特性详解，并在每个关键节点提供新旧版本的代码对比。无论你是刚接触 LangChain 的新人，还是从 0.x 迁移的老用户，都能从本文找到价值。
在深入代码之前，先理清 LangChain 1.0 的整体生态结构，这是很多入门教程忽略的重要基础。
langchain-core ← 基础抽象层（BaseMessage, BaseTool, RunnableInterface...） ↑ langchain ← 高级封装层（create_agent, Middleware, 工具链...） ↑ langchain-openai ← 模型集成层（ChatOpenAI, OpenAIEmbeddings...） langchain-anthropic ← 模型集成层（ChatAnthropic...） langchain-community ← 社区集成（各类 Retriever, VectorStore...） ↑ langchain-classic ← 向后兼容层（旧的 AgentExecutor, LCEL 等）
关键理解：
langchain-core
langchain
langchain-classic
langchain-openai
# 推荐使用 uv（更快的 pip 替代品） uv pip install --upgrade langchain langchain-openai langgraph # 如果需要旧版功能（LCEL、AgentExecutor 等） uv pip install langchain-classic # 传统方式 pip install --upgrade langchain langchain-openai langgraph
⚠️ 注意： LangChain 1.0 要求 Python 3.10+，不再支持 Python 3.9。
这是 1.0 版本最重要的架构调整之一：
重要：LangChain 的 Agent 底层就是运行在 LangGraph 之上的。 你可以先用 LangChain 的高级 API 快速开发，需要时无缝切换到 LangGraph 精细控制——两者并不对立。
import os from langchain_openai import ChatOpenAI # 也可以使用 DeepSeek（国内开发者常用） from langchain_openai import ChatOpenAI llm = ChatOpenAI( model="deepseek-chat", openai_api_key=os.getenv("DEEPSEEK_API_KEY"), openai_api_base="https://api.deepseek.com/v1", temperature=0.7, ) response = llm.invoke("你好，介绍一下 LangChain 1.0 的核心变化") print(response.content)
旧版（0.x）： 需要手动用 asyncio 或 ThreadPoolExecutor 实现并发
新版（1.0）： 内置 batch 和 batch_as_completed
batch
batch_as_completed
questions = [ "什么是 RAG？", "什么是 Agent？", "什么是 Middleware？", ] # 并发发送，等全部完成后返回 responses = llm.batch(questions) # 并发发送，有结果就立即返回（流式体验更好） for response in llm.batch_as_completed(questions): print(response.content) print("---")
适合避免 API 超限、控制成本：
from langchain_core.rate_limiters import InMemoryRateLimiter rate_limiter = InMemoryRateLimiter( requests_per_second=0.5, # 每秒最多 0.5 个请求（即每2秒1个） check_every_n_seconds=0.1, # 每0.1秒检查一次 max_bucket_size=5, # 允许短暂爆发至5个请求 ) llm = ChatOpenAI( model="gpt-4o-mini", rate_limiter=rate_limiter, )
这是 langchain-core 1.0 最重要的底层变化。旧版中，不同提供商的响应结构不统一，1.0 引入了标准化的 .content_blocks 属性：
.content_blocks
response = llm.invoke("解释一下注意力机制") # 旧版：只能访问 .content（字符串） print(response.content) # 新版：还可以访问结构化的内容块 for block in response.content_blocks: print(f"类型: {block.type}") # text, reasoning, tool_call, citation... if block.type == "text": print(f"内容: {block.text}") elif block.type == "reasoning": # Claude、DeepSeek-R1 等推理模型 print(f"推理过程: {block.thinking}")
为什么重要？ 现代 LLM 的响应不再只是纯文本，还包括： - 推理过程（reasoning traces） - 引用（citations） - 服务端工具调用（server-side tool calls）
Content Blocks 让这些跨提供商保持一致的接口。
from langchain_core.prompts import ChatPromptTemplate prompt = ChatPromptTemplate.from_messages([ ("system", "你是一个专业的{domain}领域专家，请用简洁清晰的语言回答问题。"), ("human", "{question}"), ]) # 直接格式化 formatted = prompt.format_messages( domain="电力系统", question="什么是台区线损？" ) print(formatted)
1.0 版中，create_agent 的 system_prompt 参数现在支持传入 SystemMessage 对象（不只是字符串），这在需要 cache control 等高级功能时很有用：
create_agent
system_prompt
SystemMessage
from langchain_core.messages import SystemMessage # 旧版写法（字符串） system_prompt = "你是一个智能助手" # 新版写法（SystemMessage 对象，可附加额外配置） system_message = SystemMessage( content="你是一个电力系统专家助手，擅长分析用电数据。", additional_kwargs={"cache_control": {"type": "ephemeral"}} # Anthropic cache )
from langchain_core.prompts import FewShotChatMessagePromptTemplate examples = [ { "input": "分析用户用电量突然增加50%", "output": "可能原因：1) 新增大功率设备；2) 窃电行为；3) 计量故障。建议：对比历史数据，排查异常时段。" }, { "input": "变压器三相不平衡度超过15%", "output": "需关注：1) 单相负荷分布；2) 用户用电设备类型；3) 线路接线方式。建议：调整相序分配，平衡三相负荷。" }, ] example_prompt = ChatPromptTemplate.from_messages([ ("human", "{input}"), ("ai", "{output}"), ]) few_shot_prompt = FewShotChatMessagePromptTemplate( example_prompt=example_prompt, examples=examples, ) final_prompt = ChatPromptTemplate.from_messages([ ("system", "你是电力系统故障分析专家"), few_shot_prompt, ("human", "{input}"), ])
工具是 Agent 与外部世界交互的接口，1.0 中工具定义方式进一步简化。
from langchain_core.tools import tool from typing import Annotated @tool def query_electricity_usage( user_id: Annotated[str, "用户编号，格式如 U20250001"], month: Annotated[str, "查询月份，格式 YYYY-MM"], ) -> dict: """查询指定用户在特定月份的用电量数据。""" # 实际业务中这里连接数据库 return { "user_id": user_id, "month": month, "usage_kwh": 450.5, "peak_usage": 12.3, "valley_usage": 8.7, } @tool def detect_electricity_theft( user_id: Annotated[str, "用户编号"], threshold: Annotated[float, "异常判断阈值，默认0.3"] = 0.3, ) -> dict: """对指定用户进行窃电风险评估。""" return { "user_id": user_id, "risk_score": 0.72, "risk_level": "高风险", "anomaly_features": ["深夜用电量激增", "计量误差超标"], } # 查看工具 Schema print(query_electricity_usage.args_schema.model_json_schema())
from langchain_core.tools import BaseTool from pydantic import BaseModel, Field class VideoAnalysisInput(BaseModel): camera_id: str = Field(description="摄像头编号") start_time: str = Field(description="开始时间，格式 HH:MM") end_time: str = Field(description="结束时间，格式 HH:MM") behavior_types: list[str] = Field( default=["intrusion", "loitering"], description="需要检测的行为类型" ) class VideoAnalysisTool(BaseTool): name: str = "video_behavior_analysis" description: str = "分析指定摄像头在时间段内的异常行为" args_schema: type[BaseModel] = VideoAnalysisInput def _run(self, camera_id: str, start_time: str, end_time: str, behavior_types: list[str]) -> dict: # 调用实际的视频分析 API return { "camera_id": camera_id, "events": [ {"time": "14:23", "type": "intrusion", "confidence": 0.95}, {"time": "14:31", "type": "loitering", "confidence": 0.82}, ] } video_tool = VideoAnalysisTool()
1.0 中结构化输出与 Agent 循环深度整合，消除了额外的 LLM 调用：
from langchain.agents import create_agent from langchain.agents.structured_output import ToolStrategy, ProviderStrategy from pydantic import BaseModel class ElectricityReport(BaseModel): user_id: str risk_level: str main_issues: list[str] recommendation: str confidence: float # 方式1：通过工具调用实现结构化输出（兼容所有模型） agent = create_agent( model="openai:gpt-4o-mini", tools=[query_electricity_usage, detect_electricity_theft], response_format=ToolStrategy(ElectricityReport), system_prompt="你是电力数据分析师，分析完成后输出结构化报告。", ) # 方式2：使用提供商原生结构化输出（更高效，但需模型支持） agent = create_agent( model="openai:gpt-4o-mini", tools=[query_electricity_usage, detect_electricity_theft], response_format=ProviderStrategy(ElectricityReport), system_prompt="你是电力数据分析师，分析完成后输出结构化报告。", )
这是 LangChain 1.0 变化最大、最核心的部分。
# ❌ 旧版方式一：AgentExecutor（已废弃，迁移到 langchain-classic） from langchain.agents import create_react_agent, AgentExecutor from langchain import hub prompt = hub.pull("hwchase17/react") agent = create_react_agent(llm, tools, prompt) agent_executor = AgentExecutor(agent=agent, tools=tools, verbose=True) result = agent_executor.invoke({"input": "查询用户 U001 的用电情况"}) # ❌ 旧版方式二：LangGraph prebuilt（已废弃） from langgraph.prebuilt import create_react_agent as lg_create_agent graph = lg_create_agent(llm, tools) result = graph.invoke({"messages": [("user", "查询用户 U001 的用电情况")]})
# ✅ 新版统一方式 from langchain.agents import create_agent agent = create_agent( model="openai:gpt-4o-mini", # 统一的模型字符串格式 tools=[ query_electricity_usage, detect_electricity_theft, ], system_prompt="你是国家电网的智能分析助手，帮助工作人员分析用电数据。", ) result = agent.invoke({ "role": "user", "content": "帮我查询用户 U20250001 在 2025-09 的用电情况，并评估是否有窃电风险" }) print(result)
1.0 引入了统一的 provider:model_name 格式，无需分别导入不同包：
provider:model_name
# 各提供商格式示例 "openai:gpt-4o-mini" "openai:gpt-5" "anthropic:claude-sonnet-4-5" "google-genai:gemini-2.0-flash" # DeepSeek（通过 OpenAI 兼容接口） from langchain_openai import ChatOpenAI deepseek_llm = ChatOpenAI( model="deepseek-chat", openai_api_key=os.getenv("DEEPSEEK_API_KEY"), openai_api_base="https://api.deepseek.com/v1", ) # 然后直接传入 llm 对象 agent = create_agent( model=deepseek_llm, # 也可以传入已创建的 llm 对象 tools=[...], )
理解 create_agent 的工作原理对于调试和定制至关重要：
用户输入 ↓ 【第1步】发送请求给 LLM ↓ 【第2步】LLM 决策： ├── 需要调用工具 → 执行工具 → 将结果加入对话 → 返回第1步 └── 直接回答 → 返回最终结果 ↓ 最终输出
# 异步调用（推荐生产环境） import asyncio async def analyze_user(user_id: str): agent = create_agent( model="openai:gpt-4o-mini", tools=[query_electricity_usage, detect_electricity_theft], system_prompt="你是电力数据分析师。", ) result = await agent.ainvoke({ "role": "user", "content": f"分析用户 {user_id} 的用电情况" }) return result asyncio.run(analyze_user("U20250001"))
Middleware（中间件）是 LangChain 1.0 最具创新性的特性，也是与 0.x 差异最大的地方。它解决了一个核心问题：如何在不修改核心 Agent 逻辑的情况下，对 Agent 行为进行精细化控制？
每一步 Agent 循环 = 一系列钩子（Hooks）的触发 before_agent() ← 整个 Agent 启动前 ↓ before_model() ← 每次调用 LLM 前 ↓ LLM 调用 ↓ after_model() ← 每次调用 LLM 后 ↓ before_tool() ← 每次调用工具前 ↓ Tool 执行 ↓ after_tool() ← 每次调用工具后 ↓ after_agent() ← 整个 Agent 完成后
在工具执行前暂停，等待人工审批——对于涉及敏感操作的 Agent 至关重要：
from langchain.middleware import HumanInTheLoopMiddleware # 方式1：对所有工具启用人工审批 hitl_middleware = HumanInTheLoopMiddleware() # 方式2：只对特定工具启用 hitl_middleware = HumanInTheLoopMiddleware( tools=["send_alert", "modify_user_account"] # 只有这些工具需要审批 ) agent = create_agent( model="openai:gpt-4o-mini", tools=[query_electricity_usage, detect_electricity_theft, send_alert], middleware=[hitl_middleware], system_prompt="你是电力安全巡检助手。", ) # 执行时，遇到需要审批的工具会暂停并等待确认 result = agent.invoke({"role": "user", "content": "发现高风险用户，请发送告警"})
解决长对话中 token 溢出问题，自动压缩历史消息：
from langchain.middleware import SummarizationMiddleware # 当消息超过 4000 tokens 时自动压缩历史 summarization_middleware = SummarizationMiddleware( max_tokens=4000, # 触发压缩的 token 阈值 keep_last_n=5, # 保留最近 N 条消息不压缩 summary_model="openai:gpt-4o-mini", # 用于压缩的模型 ) agent = create_agent( model="openai:gpt-4o-mini", tools=[...], middleware=[summarization_middleware], system_prompt="你是一个长会话分析助手。", )
1.1 版本更新： SummarizationMiddleware 现在可以利用 Model Profiles 自动感知模型的上下文长度，智能决定何时触发压缩。
SummarizationMiddleware
在数据发送给 LLM 之前，自动脱敏敏感信息：
from langchain.middleware import PIIRedactionMiddleware pii_middleware = PIIRedactionMiddleware( patterns=[ r'\b\d{18}\b', # 身份证号 r'\b1[3-9]\d{9}\b', # 手机号 r'\b\d{3}-\d{4}-\d{4}\b', # 固话 r'\b[A-Za-z0-9._%+-]+@[A-Za-z0-9.-]+\.[A-Z|a-z]{2,}\b', # 邮箱 ], replacement="[已脱敏]", ) agent = create_agent( model="openai:gpt-4o-mini", tools=[query_electricity_usage], middleware=[pii_middleware], system_prompt="你是用电数据分析助手。", )
实现一个记录 Agent 每步操作的审计日志 Middleware：
from langchain.middleware import BaseMiddleware from langchain_core.messages import BaseMessage from datetime import datetime import json class AuditLogMiddleware(BaseMiddleware): """记录 Agent 每一步操作的审计日志，适合电力行业合规要求""" def __init__(self, log_file: str = "agent_audit.log"): self.log_file = log_file self.session_id = None def before_agent(self, messages: list[BaseMessage], **kwargs): self.session_id = datetime.now().strftime("%Y%m%d%H%M%S%f") self._log("SESSION_START", { "session_id": self.session_id, "input_preview": str(messages[-1].content)[:100] }) return messages def before_tool(self, tool_name: str, tool_input: dict, **kwargs): self._log("TOOL_CALL", { "session_id": self.session_id, "tool": tool_name, "input": tool_input }) return tool_name, tool_input def after_tool(self, tool_name: str, tool_output: any, **kwargs): self._log("TOOL_RESULT", { "session_id": self.session_id, "tool": tool_name, "output_preview": str(tool_output)[:200] }) return tool_output def after_agent(self, output: any, **kwargs): self._log("SESSION_END", { "session_id": self.session_id, "output_preview": str(output)[:200] }) return output def _log(self, event_type: str, data: dict): log_entry = { "timestamp": datetime.now().isoformat(), "event": event_type, **data } with open(self.log_file, "a", encoding="utf-8") as f: f.write(json.dumps(log_entry, ensure_ascii=False) + "\n") # 组合多个 Middleware（按顺序执行） agent = create_agent( model="openai:gpt-4o-mini", tools=[query_electricity_usage, detect_electricity_theft], middleware=[ PIIRedactionMiddleware(patterns=[r'\b\d{18}\b']), # 先脱敏 AuditLogMiddleware(log_file="grid_agent_audit.log"), # 再记录 ], system_prompt="你是国家电网智能分析助手。", )
from langchain_core.messages import HumanMessage, AIMessage, SystemMessage # 构建对话历史 conversation_history = [ SystemMessage(content="你是电力客服助手"), HumanMessage(content="我的电费这个月为什么这么高？"), AIMessage(content="您好！电费偏高可能有以下原因：1)夏季空调用电增加 2)新增了大功率设备..."), HumanMessage(content="我家没有新增设备，应该怎么查？"), ] # Agent 会自动利用历史上下文 result = agent.invoke(conversation_history)
对于需要跨会话记忆的应用，需要借助 LangGraph 的持久化能力：
from langgraph.checkpoint.memory import MemorySaver from langchain.agents import create_agent # 创建内存检查点（生产环境替换为数据库检查点） memory = MemorySaver() agent = create_agent( model="openai:gpt-4o-mini", tools=[query_electricity_usage], system_prompt="你是电力客服助手，记住用户的历史问题。", checkpointer=memory, # 注入持久化组件 ) # 同一个 thread_id 的对话会保持上下文 config = {"configurable": {"thread_id": "user_U001_session"}} # 第一轮 agent.invoke( {"role": "user", "content": "我是 U001 用户，我的电费高吗？"}, config=config ) # 第二轮——Agent 记得第一轮 agent.invoke( {"role": "user", "content": "能帮我详细分析一下原因吗？"}, config=config )
RAG 相关组件在 1.0 中变化相对较小，但整合进 Agent 的方式更加简洁。
将检索器包装成工具，让 Agent 自主决定何时检索：
from langchain_openai import OpenAIEmbeddings from langchain_community.vectorstores import Chroma from langchain_core.tools import tool from langchain.text_splitter import RecursiveCharacterTextSplitter from langchain_community.document_loaders import DirectoryLoader # 1. 加载文档（电力规程、标准文档等） loader = DirectoryLoader("./power_grid_docs/", glob="**/*.pdf") docs = loader.load() # 2. 分块 splitter = RecursiveCharacterTextSplitter( chunk_size=500, chunk_overlap=50 ) chunks = splitter.split_documents(docs) # 3. 向量化并存储 embeddings = OpenAIEmbeddings(model="text-embedding-3-small") vectorstore = Chroma.from_documents(chunks, embeddings, persist_directory="./chroma_db") retriever = vectorstore.as_retriever(search_kwargs={"k": 5}) # 4. 包装为工具 @tool def search_power_grid_docs(query: str) -> str: """搜索电力系统技术文档和操作规程，回答技术问题。""" docs = retriever.invoke(query) return "\n\n---\n\n".join([ f"来源：{doc.metadata.get('source', '未知')}\n内容：{doc.page_content}" for doc in docs ]) # 5. 集成到 Agent rag_agent = create_agent( model="openai:gpt-4o-mini", tools=[search_power_grid_docs, query_electricity_usage], system_prompt="你是电力系统智能助手，可以查询技术文档和用电数据来回答问题。", )
# ❌ 旧版 LCEL 方式（仍可用，但不推荐新项目采用） from langchain_core.runnables import RunnablePassthrough from langchain_core.output_parsers import StrOutputParser rag_chain = ( {"context": retriever, "question": RunnablePassthrough()} | prompt | llm | StrOutputParser() ) result = rag_chain.invoke("什么是台区线损？") # ✅ 新版方式：RAG 作为工具，由 Agent 自主调用 result = rag_agent.invoke({ "role": "user", "content": "什么是台区线损？如何计算？" })
新版优势： Agent 可以根据问题类型自主决定是检索文档、查询数据库，还是直接回答，而不是每次都固定走 RAG 流程。
import asyncio async def stream_agent_response(): agent = create_agent( model="openai:gpt-4o-mini", tools=[query_electricity_usage], system_prompt="你是电力分析助手。", ) async for chunk in agent.astream({ "role": "user", "content": "帮我分析一下 2025 年的用电趋势" }): # chunk 是增量内容 print(chunk, end="", flush=True) asyncio.run(stream_agent_response())
async def stream_with_events(): agent = create_agent( model="openai:gpt-4o-mini", tools=[query_electricity_usage, detect_electricity_theft], system_prompt="你是电力分析助手。", ) async for event in agent.astream_events( {"role": "user", "content": "分析用户 U001"}, version="v2" ): event_type = event["event"] if event_type == "on_chat_model_stream": # LLM 生成文本 content = event["data"]["chunk"].content if content: print(content, end="", flush=True) elif event_type == "on_tool_start": # 工具开始执行 print(f"\n[调用工具: {event['name']}]") print(f"[参数: {event['data']['input']}]") elif event_type == "on_tool_end": # 工具执行完成 print(f"[工具结果: {event['data']['output']}]\n") asyncio.run(stream_with_events())
LangChain 1.1（2025年12月发布）在 1.0 基础上新增了 Model Profiles 系统：
from langchain_openai import ChatOpenAI llm = ChatOpenAI(model="gpt-4o-mini") # 查看模型支持的能力 profile = llm.profile print(f"支持函数调用: {profile.supports_tool_calling}") print(f"支持结构化输出: {profile.supports_structured_output}") print(f"支持推理: {profile.supports_reasoning}") print(f"最大上下文: {profile.max_context_length}") # SummarizationMiddleware 可以利用 profile 自动设置阈值 from langchain.middleware import SummarizationMiddleware smart_summarization = SummarizationMiddleware( # 不用手动设置 max_tokens，自动从 profile 读取 use_model_profile=True, keep_last_n=5, )
AgentExecutor
langgraph.prebuilt.create_react_agent
langchain.agents.create_agent
prompt \| llm \| parser
llm.batch()
llm.batch_as_completed()
InMemoryRateLimiter
.content
HumanInTheLoopMiddleware
PIIRedactionMiddleware
with_structured_output()
# 旧版代码在安装 langchain-classic 后仍可运行 pip install langchain-classic # langchain-classic 包含： # - AgentExecutor # - LCEL (|) 语法的完整支持 # - 旧版 ConversationChain # - 旧版 RetrievalQA # - create_react_agent（旧版）
迁移建议： - 新项目：直接用 1.0 API - 旧项目：先加 langchain-classic 保持运行，逐步迁移到新 API - 团队项目：优先迁移 Agent 创建部分，RAG/工具部分影响较小
将前面所有知识整合成一个完整示例：
""" 电力巡检智能助手 功能：查询用电数据、检测窃电风险、搜索规程文档、记录审计日志 """ import os import asyncio from typing import Annotated from pydantic import BaseModel from langchain_openai import ChatOpenAI from langchain.agents import create_agent from langchain.agents.structured_output import ToolStrategy from langchain.middleware import ( PIIRedactionMiddleware, SummarizationMiddleware, HumanInTheLoopMiddleware, ) from langchain_core.tools import tool from langchain_core.messages import SystemMessage from langgraph.checkpoint.memory import MemorySaver # ============= 工具定义 ============= @tool def query_electricity_usage( user_id: Annotated[str, "用户编号"], month: Annotated[str, "月份，格式 YYYY-MM"], ) -> dict: """查询用户月度用电数据""" return {"user_id": user_id, "month": month, "usage_kwh": 480.5, "status": "正常"} @tool def detect_electricity_theft( user_id: Annotated[str, "用户编号"], ) -> dict: """评估用户窃电风险""" return {"user_id": user_id, "risk_score": 0.15, "risk_level": "低风险"} @tool def send_inspection_alert( user_id: Annotated[str, "用户编号"], alert_type: Annotated[str, "告警类型"], message: Annotated[str, "告警详情"], ) -> str: """发送巡检告警（需要人工审批）""" return f"告警已发送：用户 {user_id}，类型：{alert_type}" # ============= 输出结构 ============= class InspectionReport(BaseModel): user_id: str inspection_result: str risk_level: str recommendations: list[str] requires_followup: bool # ============= Middleware 配置 ============= middlewares = [ PIIRedactionMiddleware( patterns=[r'\b\d{18}\b', r'\b1[3-9]\d{9}\b'], replacement="[已脱敏]", ), SummarizationMiddleware(max_tokens=3000, keep_last_n=5), HumanInTheLoopMiddleware(tools=["send_inspection_alert"]), ] # ============= 创建 Agent ============= memory = MemorySaver() llm = ChatOpenAI( model="deepseek-chat", openai_api_key=os.getenv("DEEPSEEK_API_KEY"), openai_api_base="https://api.deepseek.com/v1", ) agent = create_agent( model=llm, tools=[ query_electricity_usage, detect_electricity_theft, send_inspection_alert, ], middleware=middlewares, response_format=ToolStrategy(InspectionReport), system_prompt=SystemMessage(content=( "你是国家电网智能巡检助手。执行任务时请：\n" "1. 先查询用电数据，再评估风险\n" "2. 高风险（>0.7）需发送告警\n" "3. 输出结构化巡检报告\n" "4. 注意保护用户隐私" )), checkpointer=memory, ) # ============= 执行 ============= async def run_inspection(user_id: str, session_id: str): config = {"configurable": {"thread_id": session_id}} print(f"开始巡检用户 {user_id}...") async for event in agent.astream_events( {"role": "user", "content": f"请对用户 {user_id} 执行 2025-10 月份巡检"}, config=config, version="v2" ): if event["event"] == "on_tool_start": print(f" → 调用工具: {event['name']}") elif event["event"] == "on_chat_model_stream": content = event["data"]["chunk"].content if content: print(content, end="", flush=True) print("\n巡检完成") asyncio.run(run_inspection("U20250001", "session_001"))
短期： 如果维护旧项目，需要了解。 新项目： 不必学。create_agent + Middleware 覆盖了 90% 的场景，更简单、更强大。
LangChain Agent 底层就是 LangGraph，需要时可以"拆箱"直接操作底层图。
interrupt()
RAG 链的迁移是最简单的部分：把 RAG 逻辑包装成一个 @tool，然后交给 create_agent 即可。好处是 Agent 可以自主决定什么时候检索。
@tool
# 方案1：DeepSeek（最推荐，性价比高） llm = ChatOpenAI( model="deepseek-chat", # 或 deepseek-reasoner openai_api_key=os.getenv("DEEPSEEK_API_KEY"), openai_api_base="https://api.deepseek.com/v1", ) # 方案2：通义千问 from langchain_community.chat_models import ChatTongyi llm = ChatTongyi(model="qwen-max") # 方案3：智谱 GLM from langchain_community.chat_models import ChatZhipuAI llm = ChatZhipuAI(model="glm-4")
LangChain 1.0 代表了一次深刻的范式转变：
对于正在做企业 AI 应用（如 RAG 问答系统、数据分析助手、智能巡检）的开发者，现在是从 0.x 迁移到 1.0 的好时机。
参考资料： - LangChain 1.0 官方发布博文 - LangChain 官方文档 - LangChain 1.1 Changelog - LangChain Migration Guide (Python)
还没有评论，来抢沙发吧！
这个人很懒，什么都没写...
3 篇文章
Chuanyun
还没有评论，来抢沙发吧！