本文深入解析了 LangChain 的工程架构及其核心组件。内容涵盖 LangChain 的基本定义、技术栈分层（库、模板、Serve、Smith）、六大核心抽象（模型、提示、索引、记忆、链、代理）。详细阐述了从原始 API 调用到使用 SDK 的演进，重点讲解了 LCEL 表达式语言、DAG 链式调用、LangGraph 循环图结构。深入探讨了 RAG 检索增强生成的原理、关键组件及代码实现，以及 Tool 工具和 Agent 智能体的工作机制。最后补充了生产环境部署建议，包括监控、成本、延迟及安全方面的考量，为开发者构建复杂 AI 应用提供全面的技术指导。
LangChain 是一个基于开源大语言模型的 AI 工程开发框架，旨在使研究人员和开发人员能够更轻松地构建、实验和部署以自然语言处理（NLP）为中心的应用程序。它提供了多种组件和工具，可帮助用户利用最近的语言模型进展，如大型 Transformer 模型等，并且可以与 Hugging Face 等平台集成。LangChain 的核心理念是将语言模型用作协作工具，通过它，开发者可以构建出处理复杂任务的系统，并且可以高效地对接不同的数据源和应用程序接口（APIs）。
从整体架构来看，LangChain 自下而上主要包括以下核心部分：
LangChain 的核心组件和能力包括六大核心抽象，用于构建复杂的 AI 应用，同时保持了良好的扩展能力。
Models（模型）
Prompts（提示）
Indexes（索引）
Memory（记忆）
Chains（链）
Agents（代理）
总之，LangChain 是一个强大的工具箱，不仅涵盖了基础工具，还为个性化需求提供了自定义组件解决方案。它使开发者能够更专注于创新和优化产品功能，从原型到生产环境的转化变得更加高效。
直接调用 OpenAI API 需要处理大量的认证、格式化和解析工作。使用 LangChain SDK 可以大幅简化这一过程。
import os
import requests
# API Key
api_key = os.getenv('OPENAI_API_KEY')
# 头部信息
headers = {
'Content-Type': 'application/json',
'Authorization': f'Bearer {api_key}'
}
# 准备数据
data = {
'model': 'gpt-4',
'messages': [{'role': 'user', 'content': '什么是图计算？'}],
'temperature': 0.7
}
# 调用 API
url = 'https://api.openai.com/v1/chat/completions'
response = requests.post(url, json=data, headers=headers)
answer = response.json()['choices'][0]['message']['content']
print(answer)
使用 LangChain SDK 后：
from langchain_openai import ChatOpenAI
# 调用 Chat Completion API
llm = ChatOpenAI(model_name='gpt-4')
response = llm.invoke('什么是图计算？')
print(response)
对于文本生成模型服务来说，实际的输入和输出本质上都是字符串，因此直接裸调用 LLM 服务带来的问题是要在输入格式化和输出结果解析上做大量的重复的文本处理工作。LangChain 当然考虑到这一点，提供了 Prompt 和 OutputParser 抽象，用户可以根据自己的需要选择具体的实现类型使用。
from langchain_openai import ChatOpenAI
# 调用 Chat Completion API
llm = ChatOpenAI(model_name='gpt-4')
response = llm.invoke('什么是图计算？')
print(response)
结合 Prompt 和 OutputParser 的完整流程：
from langchain_core.output_parsers import StrOutputParser
from langchain_core.prompts import ChatPromptTemplate
from langchain_openai import ChatOpenAI
# 创建 LLM
llm = ChatOpenAI(model_name='gpt-4')
# 创建 Prompt
prompt = ChatPromptTemplate.from_template("{question}")
# 创建输出解析器
output_parser = StrOutputParser()
# 调用 LLM
message = prompt.invoke({'question': '什么是图计算？'})
response = llm.invoke(message)
answer = output_parser.invoke(response)
print(answer)
__or__ 运算符的思路，构建了类似 Unix 管道运算符的设计，实现更简洁的 LLM 调用形式。from langchain_core.output_parsers import StrOutputParser
from langchain_core.prompts import ChatPromptTemplate
from langchain_openai import ChatOpenAI
# 创建 LLM
llm = ChatOpenAI(model_name='gpt-4')
# 创建 Prompt
prompt = ChatPromptTemplate.from_template("{question}")
# 创建输出解析器
output_parser = StrOutputParser()
# 创建 Chain
chain = prompt | llm | output_parser
# 调用 Chain
answer = chain.invoke({'question': '什么是图计算？'})
print(answer)
from langchain_core.runnables import RunnablePassthrough
# 创建 Chain
chain = {"question": RunnablePassthrough()} | prompt | llm | output_parser
# 调用 Chain
answer = chain.invoke('什么是图计算？')
print(answer)
from operator import itemgetter
from langchain_core.output_parsers import StrOutputParser
from langchain_core.prompts import ChatPromptTemplate
from langchain_core.runnables import RunnablePassthrough
from langchain_openai import ChatOpenAI
# 创建 LLM
llm = ChatOpenAI(model_name='gpt-4')
# 创建输出解析器
output_parser = StrOutputParser()
# 创建 Prompt
topic_prompt = ChatPromptTemplate.from_template("生成一种'{input}'的名称")
good_prompt = ChatPromptTemplate.from_template("列举{topic}的好处:")
bad_prompt = ChatPromptTemplate.from_template("列举{topic}的坏处:")
summary_prompt = ChatPromptTemplate.from_messages(
[
("ai", "{topic}"),
("human", "好处:\n{good}\n\n坏处:\n{bad}"),
("system", "生成最终结论"),
]
)
# 创建组合 Chain
topic_chain = topic_prompt | llm | output_parser | {"topic": RunnablePassthrough()}
goods_chain = good_prompt | llm | output_parser
bads_chain = bad_prompt | llm | output_parser
summary_chain = summary_prompt | llm | output_parser
chain = (
topic_chain
|
{
"good": goods_chain,
"bad": bads_chain,
"topic": itemgetter("topic"),
}
| summary_chain
)
# 调用 chain
answer = chain.invoke({"input": '常见水果'})
print(answer)
from langchain_openai import ChatOpenAI
from langchain_core.messages import HumanMessage
from langgraph.graph import END, MessageGraph
# 初始化聊天模型
model = ChatOpenAI(temperature=0)
# 创建一个 MessageGraph
graph = MessageGraph()
# 添加一个名为 "oracle" 的节点，它执行聊天模型并返回结果
graph.add_node("oracle", model)
graph.add_edge("oracle", END)
graph.set_entry_point("oracle")
# 编译图
runnable = graph.compile()
# 运行图
result = runnable.invoke(HumanMessage("What is 1 + 1?"))
print(result) # 输出聊天模型的回答
通过 Chain，LangChain 相当于以'工作流'的形式，将 LLM 与 IO 组件进行了有秩序的连接，从而具备构建复杂 AI 工程流程的能力。而我们都知道 LLM 提供的文本生成服务本身不提供记忆功能，需要用户自己管理对话历史。因此引入 Memory 组件，可以很好地扩展 AI 工程的能力边界。
拥有记忆后，确实扩展了 AI 工程的应用场景。但是在专有领域，LLM 无法学习到所有的专业知识细节，因此在面向专业领域知识的提问时，无法给出可靠准确的回答，甚至会'胡言乱语'，这种现象称之为 LLM 的'幻觉'。检索增强生成（RAG）把信息检索技术和大模型结合起来，将检索出来的文档和提示词一起提供给大模型服务，从而生成更可靠的答案，有效的缓解大模型推理的'幻觉'问题。
相比提示词工程，RAG 有更丰富的上下文和数据样本，可以不需要用户提供过多的背景描述，即能生成比较符合用户预期的答案。相比于模型微调，RAG 可以提升问答内容的时效性和可靠性，同时在一定程度上保护了业务数据的隐私性。
但由于每次问答都涉及外部系统数据检索，因此 RAG 的响应时延相对较高。另外，引用的外部知识数据会消耗大量的模型 Token 资源。因此，用户需要结合自身的实际应用场景做合适的技术选型。
from langchain.text_splitter import RecursiveCharacterTextSplitter
from langchain_community.vectorstores.faiss import FAISS
from langchain_core.documents import Document
from langchain_core.output_parsers import StrOutputParser
from langchain_core.prompts import ChatPromptTemplate
from langchain_core.runnables import RunnablePassthrough
from langchain_openai import OpenAIEmbeddings, ChatOpenAI
# 创建 LLM
llm = ChatOpenAI(model_name='gpt-4')
# 创建 Prompt
prompt = ChatPromptTemplate.from_template('基于上下文：{context}\n回答：{input}')
# 创建输出解析器
output_parser = StrOutputParser()
# 模拟文档
docs = [Document(page_content="TuGraph 是蚂蚁开源的图数据库产品")]
# 文档嵌入
splits = RecursiveCharacterTextSplitter().split_documents(docs)
vector_store = FAISS.from_documents(splits, OpenAIEmbeddings())
retriever = vector_store.as_retriever()
# 创建 Chain
chain_no_context = RunnablePassthrough() | llm | output_parser
chain = (
{"context": retriever, "input": RunnablePassthrough()}
| prompt | llm | output_parser
)
# 调用 Chain
print(chain_no_context.invoke('蚂蚁图数据库开源了吗？'))
print(chain.invoke('蚂蚁图数据库开源了吗？'))
结合示例和向量数据库的存取过程，我们简单理解一下 RAG 中关键组件：
'会使用工具'是人类和动物的根本区别。
要构建更强大的 AI 工程应用，只有生成文本这样的'纸上谈兵'能力自然是不够的。工具不仅仅是'肢体'的延伸，更是为'大脑'插上了想象力的'翅膀'。借助工具，才能让 AI 应用的能力真正具备无限的可能，才能从'认识世界'走向'改变世界'。
这里不得不提到 OpenAI 的 Chat Completion API 提供的函数调用能力（注意这里不是 Assistant 的函数调用），通过在对话请求内附加 tools 参数描述工具的定义格式（原先的 functions 参数已过期），LLM 会根据提示词推断出需要调用哪些工具，并提供具体的调用参数信息。用户需要根据返回的工具调用信息，自行触发相关工具的回调。下一章内容我们可以看到工具的调用动作可以通过 Agent 自主接管。
from openai import OpenAI
import json
client = OpenAI()
# Example dummy function hard coded to return the same weather
# In production, this could be your backend API or an external API
def get_current_weather(location, unit="fahrenheit"):
"""Get the current weather in a given location"""
if "tokyo" in location.lower():
return json.dumps({"location": "Tokyo", "temperature": "10", "unit": unit})
elif "san francisco" in location.lower():
return json.dumps({"location": "San Francisco", "temperature": "72", "unit": unit})
elif "paris" in location.lower():
return json.dumps({"location": "Paris", "temperature": "22", "unit": unit})
else:
return json.dumps({"location": location, "temperature": "unknown"})
def run_conversation():
# Step 1: send the conversation and available functions to the model
messages = [{"role": "user", "content": "What's the weather like in San Francisco, Tokyo, and Paris?"}]
tools = [
{
"type": "function",
"function": {
"name": "get_current_weather",
"description": "Get the current weather in a given location",
"parameters": {
"type": "object",
"properties": {
"location": {
"type": "string",
"description": "The city and state, e.g. San Francisco, CA",
},
"unit": {"type": "string", "enum": ["celsius", "fahrenheit"]},
},
"required": ["location"],
},
},
}
]
response = client.chat.completions.create(
model="gpt-3.5-turbo-0125",
messages=messages,
tools=tools,
tool_choice="auto", # auto is default, but we'll be explicit
)
response_message = response.choices[0].message
tool_calls = response_message.tool_calls
# Step 2: check if the model wanted to call a function
if tool_calls:
# Step 3: call the function
# Note: the JSON response may not always be valid; be sure to handle errors
available_functions = {
"get_current_weather": get_current_weather,
} # only one function in this example, but you can have multiple
messages.append(response_message) # extend conversation with assistant's reply
# Step 4: send the info for each function call and function response to the model
for tool_call in tool_calls:
function_name = tool_call.function.name
function_to_call = available_functions[function_name]
function_args = json.loads(tool_call.function.arguments)
function_response = function_to_call(
location=function_args.get("location"),
unit=function_args.get("unit"),
)
messages.append(
{
"tool_call_id": tool_call.id,
"role": "tool",
"name": function_name,
"content": function_response,
}
) # extend conversation with function response
second_response = client.chat.completions.create(
model="gpt-3.5-turbo-0125",
messages=messages,
) # get a new response from the model where it can see the function response
return second_response
print(run_conversation())
使用 LangChain 定义 Tool 的简化方式：
import random
from langchain_core.output_parsers.openai_tools import JsonOutputToolsParser
from langchain_core.runnables import RunnablePassthrough
from langchain_core.tools import tool
from langchain_openai import ChatOpenAI
# 定义 Tool
@tool
def get_temperature(city: str) -> int:
"""获取指定城市的当前气温"""
return random.randint(-20, 50)
# 创建 LLM
llm = ChatOpenAI(model_name='gpt-4')
# 创建 JSON 输出解析器
output_parser = JsonOutputToolsParser()
# 创建 Chain
chain = (
RunnablePassthrough()
| llm.bind_tools(tools=[get_temperature])
| output_parser
)
# 调用 Chain
print(chain.invoke('杭州今天多少度？'))
代码示例输出：
[{'type': 'get_temperature', 'args': {'city': '杭州'}}]
Agent 的核心思想是使用大型语言模型（LLM）来选择要采取的行动序列。在 Chain 中行动序列是硬编码的，而 Agent 则采用语言模型作为推理引擎来确定以什么样的顺序采取什么样的行动。Agent 相比 Chain 最典型的特点是'自治'，它可以通过借助 LLM 专长的推理能力，自动化地决策获取什么样的知识，采取什么样的行动，直到完成用户设定的最终目标。
因此，作为一个智能体，需要具备以下核心能力：
import random
from langchain.agents import create_openai_tools_agent, \
AgentExecutor
from langchain_core.prompts import ChatPromptTemplate, MessagesPlaceholder, \
HumanMessagePromptTemplate, SystemMessagePromptTemplate
from langchain_core.tools import tool
from langchain_openai import ChatOpenAI
# 创建 LLM
llm = ChatOpenAI()
# 定义 Tool
@tool
def get_temperature(city: str) -> int:
"""获取指定城市的当前气温"""
return random.randint(-20, 50)
# 创建 Agent 提示词模板
prompt = ChatPromptTemplate.from_messages([
SystemMessagePromptTemplate.from_template('You are a helpful assistant'),
MessagesPlaceholder(variable_name='chat_history', optional=True),
HumanMessagePromptTemplate.from_template('{input}'),
MessagesPlaceholder(variable_name='agent_scratchpad')
])
# 创建 Agent
tools = [get_temperature]
agent = create_openai_tools_agent(llm, tools, prompt=prompt)
# 执行 Agent
agent_executor = AgentExecutor(agent=agent, tools=tools, verbose=True)
print(agent_executor.invoke({'input': '今天杭州多少度？'})['output'])
在实际工程中落地 LangChain 应用时，除了关注功能实现，还需考虑以下生产环境的关键因素：
监控与日志 在生产环境中，必须对 LLM 的调用进行全链路监控。建议使用 LangSmith 或类似的 APM 工具记录每个 Chain 的执行时间、Token 消耗量、错误率以及中间状态。这有助于快速定位性能瓶颈和异常行为。
成本优化 LLM 调用通常按 Token 计费。应实施缓存策略（Cache），对相同的 Prompt 和输入复用之前的结果。此外，合理选择模型版本，对于简单任务使用小参数模型，复杂任务再调用大模型，以平衡效果与成本。
延迟控制 RAG 和 Agent 模式由于涉及多次外部调用，延迟较高。可采用异步调用（Async/Await）、流式输出（Streaming）以及预取数据等手段来优化用户体验。对于实时性要求极高的场景，需评估是否必须依赖 LLM 实时推理。
安全与隐私 确保用户数据在传输和存储过程中的加密。避免将敏感信息直接传递给第三方 LLM 提供商。对于内部知识库，应实施严格的访问控制和权限管理，防止数据泄露。
综上所述，LangChain 为构建企业级 AI 应用提供了坚实的基础设施。通过合理利用其核心组件并结合最佳实践，开发者可以高效地交付稳定、可扩展且安全的智能系统。
微信公众号「极客日志」，在微信中扫描左侧二维码关注。展示文案：极客日志 zeeklog
使用加密算法（如AES、TripleDES、Rabbit或RC4）加密和解密文本明文。 在线工具，加密/解密文本在线工具，online
生成新的随机RSA私钥和公钥pem证书。 在线工具，RSA密钥对生成器在线工具，online
基于 Mermaid.js 实时预览流程图、时序图等图表，支持源码编辑与即时渲染。 在线工具，Mermaid 预览与可视化编辑在线工具，online
解析常见 curl 参数并生成 fetch、axios、PHP curl 或 Python requests 示例代码。 在线工具，curl 转代码在线工具，online
将字符串编码和解码为其 Base64 格式表示形式即可。 在线工具，Base64 字符串编码/解码在线工具，online
将字符串、文件或图像转换为其 Base64 表示形式。 在线工具，Base64 文件转换器在线工具，online