ArmNet 字元服务 文档

外观

LangChain ​

LangChain 是一个用于构建大语言模型应用的开源框架，提供了丰富的组件和工具，帮助开发者快速构建复杂的 AI 应用。

核心概念​ ​

什么是 LangChain​ ​

LangChain 提供：

  • 标准化接口： 统一的 LLM 调用方式
  • 组件库： 预构建的功能组件
  • 链式组合： 灵活的工作流编排
  • Agent 框架： 自主决策能力
  • 记忆系统： 对话状态管理

核心组件​ ​

组件说明
ModelsLLM 和聊天模型封装
Prompts提示词模板管理
Chains多步骤工作流编排
Agents自主决策和工具调用
Memory对话记忆管理
Retrievers数据检索组件
Tools外部工具集成

快速开始​ ​

安装​ ​

pip install langchain langchain-openai

基础用法​ ​

from langchain_openai import ChatOpenAI  
from langchain_core.messages import HumanMessage, SystemMessage  
  
# 初始化模型  
llm = ChatOpenAI(  
    model="gpt-4o",  
    api_key="your-api-key",  
    base_url="http://122.51.35.238:5170/v1"  
)  
  
# 简单对话  
messages = [  
    SystemMessage(content="你是一个有帮助的助手。"),  
    HumanMessage(content="什么是机器学习？")  
]  
  
response = llm.invoke(messages)  
print(response.content)

提示词模板​ ​

基础模板​ ​

from langchain_core.prompts import ChatPromptTemplate  
  
# 创建模板  
prompt = ChatPromptTemplate.from_messages([  
    ("system", "你是一个{role}专家。"),  
    ("human", "{question}")  
])  
  
# 格式化  
formatted = prompt.format_messages(  
    role="Python",  
    question="如何读取 JSON 文件？"  
)  
  
response = llm.invoke(formatted)

复杂模板​ ​

from langchain_core.prompts import PromptTemplate  
  
template = """  
你是一位{expertise}专家。  
  
用户问题：{question}  
  
请按以下格式回答：  
1. 简要回答  
2. 详细解释  
3. 示例代码（如适用）  
"""  
  
prompt = PromptTemplate(  
    input_variables=["expertise", "question"],  
    template=template  
)

链式调用（Chains）​ ​

简单链​ ​

from langchain_core.output_parsers import StrOutputParser  
  
# 使用 LCEL (LangChain Expression Language)  
chain = prompt | llm | StrOutputParser()  
  
result = chain.invoke({  
    "expertise": "Python",  
    "question": "什么是装饰器？"  
})

顺序链​ ​

from langchain_core.runnables import RunnablePassthrough  
  
# 第一步：生成代码  
code_prompt = ChatPromptTemplate.from_template(  
    "用 Python 实现以下功能：{task}"  
)  
  
# 第二步：解释代码  
explain_prompt = ChatPromptTemplate.from_template(  
    "请解释以下代码：\n{code}"  
)  
  
# 组合链  
chain = (  
    {"task": RunnablePassthrough()}  
    | code_prompt  
    | llm  
    | StrOutputParser()  
    | (lambda code: {"code": code})  
    | explain_prompt  
    | llm  
    | StrOutputParser()  
)  
  
result = chain.invoke("计算斐波那契数列")

RAG 实现​ ​

文档加载​ ​

from langchain_community.document_loaders import (  
    TextLoader,  
    PyPDFLoader,  
    WebBaseLoader  
)  
  
# 加载文本文件  
text_loader = TextLoader("document.txt")  
docs = text_loader.load()  
  
# 加载 PDF  
pdf_loader = PyPDFLoader("document.pdf")  
pdf_docs = pdf_loader.load()  
  
# 加载网页  
web_loader = WebBaseLoader("https://example.com")  
web_docs = web_loader.load()

文本分割​ ​

from langchain.text_splitter import RecursiveCharacterTextSplitter  
  
splitter = RecursiveCharacterTextSplitter(  
    chunk_size=1000,  
    chunk_overlap=200,  
    separators=["\n\n", "\n", "。", ".", " ", ""]  
)  
  
chunks = splitter.split_documents(docs)

向量存储​ ​

from langchain_openai import OpenAIEmbeddings  
from langchain_community.vectorstores import Chroma  
  
# 创建嵌入模型  
embeddings = OpenAIEmbeddings(  
    model="text-embedding-3-small",  
    api_key="your-api-key",  
    base_url="http://122.51.35.238:5170/v1"  
)  
  
# 创建向量存储  
vectorstore = Chroma.from_documents(  
    documents=chunks,  
    embedding=embeddings,  
    persist_directory="./chroma_db"  
)  
  
# 创建检索器  
retriever = vectorstore.as_retriever(  
    search_type="similarity",  
    search_kwargs={"k": 5}  
)

完整 RAG 链​ ​

from langchain_core.runnables import RunnableParallel  
  
# RAG 提示词  
rag_prompt = ChatPromptTemplate.from_template("""  
基于以下信息回答问题。如果信息中没有相关内容，请说明。  
  
信息：  
{context}  
  
问题：{question}  
""")  
  
# 构建 RAG 链  
rag_chain = (  
    RunnableParallel({  
        "context": retriever | (lambda docs: "\n\n".join(d.page_content for d in docs)),  
        "question": RunnablePassthrough()  
    })  
    | rag_prompt  
    | llm  
    | StrOutputParser()  
)  
  
# 使用  
answer = rag_chain.invoke("什么是机器学习？")

Agent​ ​

基础 Agent​ ​

from langchain.agents import create_openai_tools_agent, AgentExecutor  
from langchain.tools import tool  
  
# 定义工具  
@tool  
def search_web(query: str) -> str:  
    """搜索网络获取信息"""  
    # 实际的搜索实现  
    return f"搜索结果：{query}"  
  
@tool  
def calculate(expression: str) -> str:  
    """计算数学表达式"""  
    try:  
        return str(eval(expression))  
    except:  
        return "计算错误"  
  
tools = [search_web, calculate]  
  
# 创建 Agent  
agent_prompt = ChatPromptTemplate.from_messages([  
    ("system", "你是一个有帮助的助手，可以使用工具来帮助用户。"),  
    ("human", "{input}"),  
    ("placeholder", "{agent_scratchpad}")  
])  
  
agent = create_openai_tools_agent(llm, tools, agent_prompt)  
agent_executor = AgentExecutor(agent=agent, tools=tools, verbose=True)  
  
# 运行  
result = agent_executor.invoke({"input": "计算 123 * 456"})

ReAct Agent​ ​

from langchain.agents import create_react_agent  
from langchain import hub  
  
# 使用预定义的 ReAct 提示词  
react_prompt = hub.pull("hwchase17/react")  
  
react_agent = create_react_agent(llm, tools, react_prompt)  
react_executor = AgentExecutor(agent=react_agent, tools=tools, verbose=True)  
  
result = react_executor.invoke({"input": "北京今天天气怎么样？"})

记忆系统​ ​

对话记忆​ ​

from langchain.memory import ConversationBufferMemory  
from langchain.chains import ConversationChain  
  
memory = ConversationBufferMemory()  
  
conversation = ConversationChain(  
    llm=llm,  
    memory=memory,  
    verbose=True  
)  
  
# 多轮对话  
conversation.predict(input="我叫小明")  
conversation.predict(input="我刚才说我叫什么？")

摘要记忆​ ​

from langchain.memory import ConversationSummaryMemory  
  
summary_memory = ConversationSummaryMemory(llm=llm)

向量记忆​ ​

from langchain.memory import VectorStoreRetrieverMemory  
  
# 使用向量存储的长期记忆  
vector_memory = VectorStoreRetrieverMemory(  
    retriever=vectorstore.as_retriever()  
)

输出解析​ ​

结构化输出​ ​

from langchain_core.output_parsers import JsonOutputParser  
from pydantic import BaseModel, Field  
  
class Person(BaseModel):  
    name: str = Field(description="姓名")  
    age: int = Field(description="年龄")  
    occupation: str = Field(description="职业")  
  
parser = JsonOutputParser(pydantic_object=Person)  
  
prompt = ChatPromptTemplate.from_messages([  
    ("system", "提取以下文本中的人物信息。{format_instructions}"),  
    ("human", "{text}")  
]).partial(format_instructions=parser.get_format_instructions())  
  
chain = prompt | llm | parser  
  
result = chain.invoke({"text": "张三，25岁，是一名软件工程师。"})  
# result = {"name": "张三", "age": 25, "occupation": "软件工程师"}

回调和监控​ ​

from langchain_core.callbacks import BaseCallbackHandler  
  
class CustomCallback(BaseCallbackHandler):  
    def on_llm_start(self, serialized, prompts, **kwargs):  
        print(f"开始调用 LLM...")  
      
    def on_llm_end(self, response, **kwargs):  
        print(f"LLM 调用完成，Token: {response.llm_output}")  
      
    def on_llm_error(self, error, **kwargs):  
        print(f"LLM 错误: {error}")  
  
# 使用回调  
llm_with_callback = ChatOpenAI(  
    model="gpt-4o",  
    callbacks=[CustomCallback()]  
)

最佳实践​ ​

1. 使用 LCEL​ ​

优先使用 LangChain Expression Language：

  • 更简洁的语法
  • 更好的类型支持
  • 自动支持流式和批处理

2. 模块化设计​ ​

# 将复杂链拆分为可复用的组件  
summarizer = prompt_summarize | llm | StrOutputParser()  
translator = prompt_translate | llm | StrOutputParser()  
  
# 组合使用  
full_chain = summarizer | translator

3. 错误处理​ ​

from langchain_core.runnables import RunnableWithFallbacks  
  
# 添加降级模型  
chain_with_fallback = chain.with_fallbacks([  
    fallback_chain  
])

相关资源​ ​

本页内容来自文档知识库整理，已按 ArmNet 字元服务 服务命名统一更新。