当前位置： 首页 > news >正文

Agent应用开发相关知识梳理——1.LangChain框架理解

news 2026/3/29 20:10:41

一. LangChain的本质

1.LangChain的作用是，为了更简单，便捷，规范的开发Agent模型。LangChain类似于Pythorch，本质是将很多代码逻辑写好，封装起来，当导入框架后，直接调用相关函数即可实现相关功能。

2.单纯调用LLM接口，只能达到一问一答的效果；但是使用LangChain框架，就可以将LLM API编排成具有多步骤记忆，能够去调用外部API实现更多功能逻辑的复杂AI应用。

LangChain包含多个生态，比如：

├─ langchain-core （核心零件，必装）

├─ langchain-openai （OpenAI专用零件）

├─ langchain-community （社区贡献的零件）

├─ langchain （主程序，组装工具）

├─ langgraph （高级组装工具）

└─ langsmith （调试工具）

二. LangChain的功能模块

模块1：Model（调用LLM）

from langchain_openai import ChatOpenAI llm = ChatOpenAI(model="gpt-4o")

1.单次调用

response = llm.invoke("写一首唐诗") print(response.content) 输出例如：床前明月光，疑是地上霜。举头望明月，低头思故乡。

需要等LLM把内容生成完，最后一次性返回结果；适合后台处理。

2.流式输出

for chunk in llm.stream("写一首唐诗"): print(chunk.content, end="", flush=True) 输出类似：床前明→床前明月→床前明月光→床前明月光，→...

边生成边返回；适合实时交互。

模块2：Prompt Template（提示词模板）

模版存在的原因是为了减少重复性的编码工作。

不用模版（硬编码）： llm.invoke("你是翻译官，把'hello'翻译成中文") llm.invoke("你是翻译官，把'world'翻译成中文") # 重复写"你是翻译官" 用模版（变量化）： from langchain_core.prompts import ChatPromptTemplate prompt = ChatPromptTemplate.from_messages([ ("system", "你是{role}"), # 限制AI身份/规则 ("human", "{question}") # 用户输入 ]) # 重复使用，只需改变量 prompt.invoke({"role": "翻译官", "question": "hello"}) prompt.invoke({"role": "诗人", "question": "写首诗"})

模块3：Output Parser（输出解析）

1.返回纯文本

from langchain_core.output_parsers import StrOutputParser parser = StrOutputParser() result = parser.invoke(llm_response) # 提取 response.content print(result) # "你好！我是AI助手。"

2.返回结构化数据

from pydantic import BaseModel class MovieInfo(BaseModel): title: str # 电影名 year: int # 上映年份 score: float # 评分 # 让LLM输出符合这个结构 structured_llm = llm.with_structured_output(MovieInfo) result = structured_llm.invoke("介绍一下《盗梦空间》") print(result.title) # "盗梦空间" print(result.year) # 2010 print(result.score) # 9.3

模块4：Chain（LCEL 管道语法）

chain = prompt | llm | StrOutputParser() # |：数据从左往右流动，像工厂流水线。

如果不使用管道，则需要多步调用，例如：

# 不用管道的写法（笨拙） step1 = prompt.invoke({"role": "翻译官", "question": "hello"}) step2 = llm.invoke(step1) step3 = StrOutputParser().invoke(step2)

使用管道，一步调用即可，例如：

from langchain_core.prompts import ChatPromptTemplate from langchain_openai import ChatOpenAI from langchain_core.output_parsers import StrOutputParser # 三个零件 prompt = ChatPromptTemplate.from_messages([ ("system", "你是{role}"), ("human", "{question}") ]) llm = ChatOpenAI(model="gpt-4o") parser = StrOutputParser() # 用 | 串成链 chain = prompt | llm | parser # 一次调用走完三步 result = chain.invoke({"role": "翻译官", "question": "hello"}) print(result) # "你好" #数据流向： {"role": "翻译官", "question": "hello"} ↓ (prompt组装) [系统: 你是翻译官 / 用户: hello] ↓ (llm调用) AIMessage(content="你好") ↓ (parser提取) "你好"

模块5：Memory（多轮对话记忆）

没有Memory：

chain.invoke({"input": "我叫张三"}) # "你好，张三" chain.invoke({"input": "我叫什么名字？"}) # "我不知道你的名字"

第二次调用时，LLM 忘了第一次的对话。

使用Memory：

在第一次调用后，信息会被记录起来，用字典等形式；后续调用则会记忆之前调用内容。

from langchain_community.chat_message_histories import ChatMessageHistory from langchain_core.runnables.history import RunnableWithMessageHistory # ① 存储历史消息的字典 store = {} def get_session_history(session_id: str): if session_id not in store: store[session_id] = ChatMessageHistory() return store[session_id] # ② 定义prompt（注意 {history} 占位符） prompt = ChatPromptTemplate.from_messages([ ("system", "你是助手"), ("placeholder", "{history}"), # ← 历史消息插入点 ("human", "{input}") ]) chain = prompt | llm | StrOutputParser() # ③ 给chain加上记忆能力 chain_with_memory = RunnableWithMessageHistory( chain, get_session_history, # 根据session_id查历史 input_messages_key="input", # 用户输入的key history_messages_key="history", # 历史消息的key ) # ④ 调用时带上 session_id chain_with_memory.invoke( {"input": "我叫张三"}, config={"configurable": {"session_id": "user_001"}} ) # 第二次调用 chain_with_memory.invoke( {"input": "我叫什么？"}, config={"configurable": {"session_id": "user_001"}} ) # "你叫张三"
  1. 第一次调用后，ChatMessageHistory记录了[("human", "我叫张三"), ("ai", "你好张三")]
  2. 第二次调用时，把历史消息插入{history}占位符
  3. LLM 看到完整上下文：
    [系统] 你是助手 [历史] 用户：我叫张三 / AI：你好张三 [当前] 用户：我叫什么？

模块6：RAG（检索增强生成）

如果缺少RAG，那么模型只是记得训练时学到的内容。RAG相当于给大模型更加精准的业务数据，使模型根据Prompt需求，更精准的检索出所需内容。

# ===== ① 加载文档 ===== from langchain_community.document_loaders import PyPDFLoader docs = PyPDFLoader("财报.pdf").load() # docs = [Document(page_content="2024Q1营收10亿", metadata={...}), ...] # ===== ② 切成小块（因为文档太长，LLM装不下）===== from langchain_text_splitters import RecursiveCharacterTextSplitter splitter = RecursiveCharacterTextSplitter(chunk_size=500) chunks = splitter.split_documents(docs) # ===== ③ 存入向量数据库 ===== from langchain_openai import OpenAIEmbeddings from langchain_community.vectorstores import FAISS embeddings = OpenAIEmbeddings() # 把文本转成向量 vectorstore = FAISS.from_documents(chunks, embeddings) retriever = vectorstore.as_retriever(search_kwargs={"k": 3}) # 每次检索3条最相关 # ===== ④ 检索 + 生成 ===== from langchain_core.runnables import RunnablePassthrough prompt = ChatPromptTemplate.from_messages([ ("system", "根据以下资料回答：\n{context}"), ("human", "{question}") ]) rag_chain = ( { "context": retriever, # 根据question自动检索相关文档 "question": RunnablePassthrough() # 原样传递用户问题 } | prompt | llm | StrOutputParser() ) answer = rag_chain.invoke("2024Q1营收是多少？")

模块7：Tool & Agent（LLM 自主调用工具）

Tool相当于给LLM模型提供的函数。
from langchain_core.tools import tool @tool def get_weather(city: str) -> str: """查询指定城市的天气""" # ← LLM读这行来理解工具用途 # 实际调用天气API return f"{city}：晴天，25°C"
Agent相当于LLM模型自己决定调用什么工具。
from langgraph.prebuilt import create_react_agent from langchain_core.tools import tool @tool def calculator(expression: str) -> str: """计算数学表达式""" return str(eval(expression)) @tool def get_weather(city: str) -> str: """查询天气""" return f"{city}：晴" # 创建Agent agent = create_react_agent( model=llm, tools=[calculator, get_weather] # 给LLM配备工具 ) # 调用 agent.invoke({"messages": [("human", "北京天气如何？顺便帮我算 123*456")]}) # Agent 内部流程（ReAct循环）： ① LLM思考："用户要查天气 + 算数，我需要调两个工具" ② 调用 get_weather("北京") → "北京：晴" ③ 调用 calculator("123*456") → "56088" ④ LLM汇总："北京今天晴，123*456=56088"

模块8：LangGraph（画流程图式编排）

Chain的问题：只能线性流水线A → B → C；而LangGraph的能力：可以画复杂流程图。

from langgraph.graph import StateGraph, START, END from typing import TypedDict # ① 定义状态（全局数据） class State(TypedDict): input: str output: str # ② 定义节点（每个节点是一个函数） def process_node(state: State): result = llm.invoke(state["input"]) return {"output": result.content} # ③ 画流程图 graph = StateGraph(State) graph.add_node("processor", process_node) # 添加节点 graph.add_edge(START, "processor") # 起点 → processor graph.add_edge("processor", END) # processor → 终点 app = graph.compile() # ④ 运行 result = app.invoke({"input": "你好"}) print(result["output"])

三. 企业如何决策？

阶段目标用什么举例
PoC验证证明"AI能干这事"prompt | llm | parserJupyter里跑个demo
MVP内部团队能用LangGraph + FastAPI + Redis内网聊天机器人
生产线上服务用户上面的 + 监控/缓存/备用模型客服系统
  • PoC → MVP：从"能跑"到"稳定跑"，加错误处理、记忆持久化
  • MVP → 生产：从"内部用"到"对外服务"，加监控、降本、安全审查
http://www.jsqmd.com/news/556282/

相关文章：

  • DAMOYOLO-S快速部署：GPU实例选择建议与显存占用实测数据
  • Python恶搞神器：用tkinter和threading打造随机位置无限弹窗
  • 如何用Qwen3-ASR-1.7B为视频自动生成字幕？实战教程来了
  • KS-Downloader：快手无水印内容获取工具全解析
  • 最强翻译模型Hunyuan-MT-7B一键部署：5分钟搞定33种语言互译
  • TrollInstallerX深度解析：iOS 14.0-16.6.1设备上的TrollStore安装实战指南
  • Music-dl实战指南：多平台音乐下载工具的高效部署与优化方案
  • Vue3下拉刷新组件实战：从零封装到全局注册（附完整代码）
  • LeetCode 2839. 判断通过操作能否让字符串相等 I（Python）超详细题解｜贪心算法+模拟
  • Jimeng AI Studio Z-Image Turbo部署教程：A10/A100云服务器高性能配置
  • Equalizer APO：3个步骤让Windows音频效果提升200%
  • 网盘直链解析引擎：打破下载速度壁垒的技术方案
  • etcd和brpc的联合运作在即使通讯系统中的原理
  • Windows 环境下利用 nmap 进行 UDP 端口连通性测试实战
  • PostgresSQL 更改数据库存储目录
  • uni-app微信小程序版本更新策略：冷启动与热启动的优化实践
  • 摊铺机液压系统及组件系列图
  • OBS Studio架构深度解析：如何构建专业级直播系统的核心技术栈
  • Nano-Banana软萌拆拆屋效果展示：潜水服密封拉链结构图
  • 掌控 OpenClaw：核心命令行
  • 轻量级向量引擎：SQLite-vec赋能Java应用的极简方案
  • 3/15
  • 从零构建SST实验范式：基于PsychoPy的抑制控制测量实践
  • 从素材到成片：AI 一站式极速输出——影视创作的新时代革命
  • ARMv8-A中断处理避坑指南：GICv3配置与多核中断路由那些事儿
  • 梳理资料分析的相关知识点
  • 时间管理大师：OpenClaw+nanobot自动规划每日日程
  • 嵌入式Linux开发板CH340驱动安装避坑指南（附详细步骤图）
  • XTDrone室内三维重建实战：如何用Realsense深度相机与VINS-Fusion+RTABMap构建稠密地图
  • 容器镜像瘦身实战：从1GB压到20MB，多阶段构建+清理+distroless完整流程