LangChain 工程架构深度解析

使用 LangChain SDK 后：

from langchain_openai import ChatOpenAI

# 调用 Chat Completion API
llm = ChatOpenAI(model_name='gpt-4')
response = llm.invoke('什么是图计算？')
print(response)

IO 抽象

对于文本生成模型服务来说，实际的输入和输出本质上都是字符串，因此直接裸调用 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)

组装成链

1. LCEL (LangChain Expression Language) LangChain 的表达式语言（LCEL）通过重载 __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)

2. RunnablePassthrough 为了简化 Chain 的参数调用格式，也可以借助 RunnablePassthrough 透传上游参数输入。

from langchain_core.runnables import RunnablePassthrough

# 创建 Chain
chain = {"question": RunnablePassthrough()} | prompt | llm | output_parser

# 调用 Chain
answer = chain.invoke('什么是图计算？')
print(answer)

3. DAG (有向无环图) 代码示例展示了如何通过 DAG 结构并行处理多个子任务并汇总结果。

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)

4. LangGraph 基于 LCEL 确实能描述比较复杂的 LangChain 计算图结构，但依然有 DAG 天然的设计限制，即不能支持'循环'。于是 LangChain 社区推出了一个新的项目——LangGraph，期望基于 LangChain 构建支持循环和跨多链的计算图结构，以描述更复杂的，甚至具备自动化属性的 AI 工程应用逻辑，比如智能体应用。

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)  # 输出聊天模型的回答

Memory（记忆）

通过 Chain，LangChain 相当于以'工作流'的形式，将 LLM 与 IO 组件进行了有秩序的连接，从而具备构建复杂 AI 工程流程的能力。而我们都知道 LLM 提供的文本生成服务本身不提供记忆功能，需要用户自己管理对话历史。因此引入 Memory 组件，可以很好地扩展 AI 工程的能力边界。

RAG（检索增强）

拥有记忆后，确实扩展了 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 中关键组件：

• DocumentLoader：从外部系统检索文档数据。简单起见，示例中直接构造了测试文档对象。实际上 LangChain 提供了文档加载器 BaseLoader 的接口抽象和大量实现，具体可根据自身需要选择使用。
• TextSplitter：将文档分割成块，以适应大模型上下文窗口。示例中采用了常用的 RecursiveCharacterTextSplitter，其他参考 LangChain 的 TextSplitter 接口和实现。
• EmbeddingsModel：文本嵌入模型，提供将文本编码为向量的能力。文档写入和查询匹配前都会先执行文本嵌入编码。示例采用了 OpenAI 的文本嵌入模型服务，其他参考 LangChain 的 Embeddings 接口和实现。
• VectorStore：向量存储，提供向量存储和相似性检索（ANN 算法）能力。LangChain 支持的向量存储参考 VectorStore 接口和实现。示例采用了 Meta 的 Faiss 向量数据库。
• Retriever：向量存储的查询器。一般和 VectorStore 配套实现，通过 as_retriever 方法获取，LangChain 提供的 Retriever 抽象接口是 BaseRetriever。

Tool（插件）

'会使用工具'是人类和动物的根本区别。

要构建更强大的 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（智能体）

Agent 的核心思想是使用大型语言模型（LLM）来选择要采取的行动序列。在 Chain 中行动序列是硬编码的，而 Agent 则采用语言模型作为推理引擎来确定以什么样的顺序采取什么样的行动。Agent 相比 Chain 最典型的特点是'自治'，它可以通过借助 LLM 专长的推理能力，自动化地决策获取什么样的知识，采取什么样的行动，直到完成用户设定的最终目标。

因此，作为一个智能体，需要具备以下核心能力：

• 规划：借助于 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 应用时，除了关注功能实现，还需考虑以下生产环境的关键因素：

1. 监控与日志 在生产环境中，必须对 LLM 的调用进行全链路监控。建议使用 LangSmith 或类似的 APM 工具记录每个 Chain 的执行时间、Token 消耗量、错误率以及中间状态。这有助于快速定位性能瓶颈和异常行为。

2. 成本优化 LLM 调用通常按 Token 计费。应实施缓存策略（Cache），对相同的 Prompt 和输入复用之前的结果。此外，合理选择模型版本，对于简单任务使用小参数模型，复杂任务再调用大模型，以平衡效果与成本。

3. 延迟控制 RAG 和 Agent 模式由于涉及多次外部调用，延迟较高。可采用异步调用（Async/Await）、流式输出（Streaming）以及预取数据等手段来优化用户体验。对于实时性要求极高的场景，需评估是否必须依赖 LLM 实时推理。

4. 安全与隐私 确保用户数据在传输和存储过程中的加密。避免将敏感信息直接传递给第三方 LLM 提供商。对于内部知识库，应实施严格的访问控制和权限管理，防止数据泄露。

综上所述，LangChain 为构建企业级 AI 应用提供了坚实的基础设施。通过合理利用其核心组件并结合最佳实践，开发者可以高效地交付稳定、可扩展且安全的智能系统。