LangChain：LLM 应用开发框架

# LangChain 核心组件一览
from langchain_core.prompts import ChatPromptTemplate
from langchain_core.output_parsers import StrOutputParser
from langchain_openai import ChatOpenAI

# 组装最简单的 LLM 管线
prompt = ChatPromptTemplate.from_template("说说{topic}的三个关键点")
model = ChatOpenAI(model="gpt-4")
parser = StrOutputParser()

# 使用 | 操作符链接组件（Expression Language）
chain = prompt | model | parser
result = chain.invoke({"topic": "机器学习"})
print(result)

// TypeScript 版本的 LangChain
import { ChatOpenAI } from "@langchain/openai";
import { ChatPromptTemplate } from "@langchain/core/prompts";
import { StringOutputParser } from "@langchain/core/output_parsers";

const prompt = ChatPromptTemplate.fromTemplate(
  "说说{topic}的三个关键点"
);
const model = new ChatOpenAI({ model: "gpt-4" });
const parser = new StringOutputParser();

const chain = prompt.pipe(model).pipe(parser);
const result = await chain.invoke({ topic: "机器学习" });
console.log(result);

from langchain_core.prompts import (
    ChatPromptTemplate,
    MessagesPlaceholder,
    FewShotChatMessagePromptTemplate,
)

# 带角色和占位符的模板
prompt = ChatPromptTemplate.from_messages([
    ("system", "你是一个专业的{domain}工程师，用简洁的语言回答。"),
    MessagesPlaceholder("history"),
    ("human", "{question}"),
])

# Few-shot 示例引导
examples = [
    {"input": "2+2=?", "output": "4"},
    {"input": "3*5=?", "output": "15"},
]
few_shot = FewShotChatMessagePromptTemplate(
    examples=examples,
    example_prompt=ChatPromptTemplate.from_messages([
        ("human", "{input}"),
        ("ai", "{output}"),
    ]),
)
final_prompt = ChatPromptTemplate.from_messages([
    ("system", "你是一个数学助手。"),
    few_shot,
    ("human", "{question}"),
])

# 结构化输出：强制模型返回 JSON
from langchain_core.output_parsers import JsonOutputParser
from langchain_core.pydantic_v1 import BaseModel, Field

class MovieReview(BaseModel):
    title: str = Field(description="电影标题")
    rating: float = Field(description="评分 1-10", ge=1, le=10)
    pros: list[str] = Field(description="优点列表")
    cons: list[str] = Field(description="缺点列表")

parser = JsonOutputParser(pydantic_object=MovieReview)
prompt = ChatPromptTemplate.from_messages([
    ("system", "你是一个影评专家。\n{format_instructions}"),
    ("human", "请评论电影：{movie_name}"),
]).partial(format_instructions=parser.get_format_instructions())

# 模板 + 模型 + 解析器
chain = prompt | ChatOpenAI() | parser
result = chain.invoke({"movie_name": "星际穿越"})

from langchain_core.runnables import RunnableParallel, RunnablePassthrough, RunnableBranch

# 并行执行：同时生成摘要和翻译
model = ChatOpenAI(model="gpt-4")

branch_chain = RunnableParallel(
    summary=RunnablePassthrough() | ChatPromptTemplate.from_template(
        "用一句话总结：{text}"
    ) | model,
    translation=RunnablePassthrough() | ChatPromptTemplate.from_template(
        "翻译成英文：{text}"
    ) | model,
)

result = branch_chain.invoke({
    "text": "LangChain 是一个强大的 LLM 应用框架。"
})
print("摘要:", result["summary"].content)
print("翻译:", result["translation"].content)

# 条件分支：根据问题类型路由到不同处理管线
from langchain_core.runnables import RunnableBranch

math_chain = ChatPromptTemplate.from_template("计算：{q}") | model
coding_chain = ChatPromptTemplate.from_template("写代码：{q}") | model
general_chain = ChatPromptTemplate.from_template("回答：{q}") | model

router = RunnableBranch(
    (lambda x: "计算" in x["q"], math_chain),
    (lambda x: "代码" in x["q"] or "写" in x["q"], coding_chain),
    general_chain,  # 默认分支
)

# 自动路由到正确的处理管线
result = router.invoke({"q": "计算 1+2+3"})

from langchain.memory import (
    ConversationBufferWindowMemory,
    ConversationSummaryMemory,
)
from langchain_openai import ChatOpenAI

# 滑动窗口记忆：只保留最近 3 轮对话
window_memory = ConversationBufferWindowMemory(
    k=3,  # 最近 3 轮
    return_messages=True,
    memory_key="history",
)

# 摘要记忆：自动将历史对话压缩为摘要
summary_memory = ConversationSummaryMemory(
    llm=ChatOpenAI(model="gpt-4"),
    return_messages=True,
    memory_key="summary",
)

# 在 Chain 中使用
from langchain.chains import ConversationChain

chain = ConversationChain(
    llm=ChatOpenAI(model="gpt-4"),
    memory=window_memory,
    verbose=True,
)
response = chain.predict(input="我们刚才讨论了什么？")

# 自定义 Memory：结合向量数据库实现长期记忆
from langchain.memory import VectorStoreRetrieverMemory
from langchain.vectorstores import FAISS
from langchain.embeddings import OpenAIEmbeddings
from langchain.docstore import InMemoryDocstore

# 创建向量存储
embeddings = OpenAIEmbeddings()
vectorstore = FAISS(
    embedding_function=embeddings,
    index=None,
    docstore=InMemoryDocstore(),
    index_to_docstore_id={},
)

# 向量检索记忆：按语义相似度检索相关历史
retriever_memory = VectorStoreRetrieverMemory(
    retriever=vectorstore.as_retriever(
        search_type="similarity",
        search_kwargs={"k": 3}
    ),
    memory_key="relevant_context",
)

# 保存记忆
retriever_memory.save_context(
    {"input": "我喜欢吃川菜"},
    {"output": "好的，我记住了你喜欢川菜。"}
)

from langchain.tools import tool
from langchain_openai import ChatOpenAI

# 使用装饰器快速定义工具
@tool
def calculate_age(birth_year: int) -> int:
    """根据出生年份计算年龄（2026年为基准）。"""
    return 2026 - birth_year

@tool
def get_weather(city: str) -> str:
    """查询指定城市的当前天气情况。"""
    # 实际项目中调用天气 API
    weather_db = {"北京": "晴 25°C", "上海": "多云 28°C"}
    return weather_db.get(city, f"未知城市: {city}")

# 将工具绑定到 LLM
llm = ChatOpenAI(model="gpt-4")
llm_with_tools = llm.bind_tools([calculate_age, get_weather])

# 模型自动决定何时调用工具
response = llm_with_tools.invoke("北京今天天气怎么样？我1990年出生，今年多大？")

# 结构化工具定义 + 手动执行
from langchain_core.tools import StructuredTool
from pydantic import BaseModel, Field

class SearchInput(BaseModel):
    query: str = Field(description="搜索关键词")
    max_results: int = Field(description="最大返回数", default=5)

def search_wikipedia(query: str, max_results: int = 5) -> str:
    """从维基百科搜索信息。"""
    import wikipedia
    results = wikipedia.search(query, results=max_results)
    summaries = [wikipedia.summary(r, sentences=2) for r in results[:2]]
    return "\n".join(summaries)

search_tool = StructuredTool.from_function(
    func=search_wikipedia,
    name="wikipedia_search",
    description="搜索维基百科获取背景知识",
    args_schema=SearchInput,
    return_direct=False,
)

from langchain.agents import create_openai_functions_agent, AgentExecutor
from langchain_core.prompts import ChatPromptTemplate, MessagesPlaceholder
from langchain_openai import ChatOpenAI
from langchain.tools import tool

@tool
def search(query: str) -> str:
    """搜索互联网获取信息。"""
    return f"搜索结果: 关于'{query}'的相关信息..."

@tool
def calculator(expression: str) -> str:
    """计算数学表达式。"""
    return str(eval(expression))

# 创建 Agent
llm = ChatOpenAI(model="gpt-4", temperature=0)
tools = [search, calculator]

prompt = ChatPromptTemplate.from_messages([
    ("system", "你是一个有帮助的助手，善用工具来回答问题。"),
    ("human", "{input}"),
    MessagesPlaceholder("agent_scratchpad"),
])

agent = create_openai_functions_agent(llm, tools, prompt)
executor = AgentExecutor(
    agent=agent,
    tools=tools,
    verbose=True,
    max_iterations=5,
    handle_parsing_errors=True,
)

result = executor.invoke({"input": "今天北京天气如何？"})

# 自定义 Agent：实现 ReAct 推理循环
from langchain.agents import AgentOutputParser
from langchain_core.agents import AgentAction, AgentFinish
import re

class ReactOutputParser(AgentOutputParser):
    """解析 ReAct 格式的输出。"""

    def parse(self, text: str) -> AgentAction | AgentFinish:
        # 检查是否有 Action
        if "Action:" in text:
            action_match = re.search(r"Action:\s*(.+)", text)
            input_match = re.search(r"Action Input:\s*(.+)", text, re.DOTALL)
            if action_match and input_match:
                return AgentAction(
                    tool=action_match.group(1).strip(),
                    tool_input=input_match.group(1).strip(),
                    log=text,
                )
        # 最终答案
        if "Final Answer:" in text:
            answer = text.split("Final Answer:")[-1].strip()
            return AgentFinish({"output": answer}, log=text)
        raise ValueError(f"无法解析: {text}")

from langchain.document_loaders import PyPDFLoader
from langchain.text_splitter import RecursiveCharacterTextSplitter
from langchain.embeddings import OpenAIEmbeddings
from langchain.vectorstores import FAISS
from langchain.chains import RetrievalQA
from langchain_openai import ChatOpenAI

# 1. 加载文档
loader = PyPDFLoader("langchain_docs.pdf")
documents = loader.load()

# 2. 分块（递归字符分割）
splitter = RecursiveCharacterTextSplitter(
    chunk_size=500,      # 每块 500 字符
    chunk_overlap=50,    # 重叠 50 字符保持上下文
    separators=["\n\n", "\n", "。", "，", " ", ""],
)
chunks = splitter.split_documents(documents)

# 3. 向量化并存储
embeddings = OpenAIEmbeddings()
vectorstore = FAISS.from_documents(chunks, embeddings)

# 4. 检索 + 生成
retriever = vectorstore.as_retriever(search_kwargs={"k": 3})
qa_chain = RetrievalQA.from_chain_type(
    llm=ChatOpenAI(model="gpt-4"),
    retriever=retriever,
    return_source_documents=True,
)

# 进阶 RAG：带上下文压缩和重排序
from langchain.chains import ContextualCompressionRetrieval
from langchain.retrievers import ContextualCompressionRetriever
from langchain.retrievers.document_compressors import LLMChainExtractor
from langchain_core.prompts import PromptTemplate

# 自定义 Prompt 模板
RAG_PROMPT = PromptTemplate.from_template("""基于以下上下文回答问题。如果上下文中没有相关信息，请直接说明。

上下文：
{context}

问题：{question}
回答：""")

# 上下文压缩：用 LLM 提取每个文档中与问题相关的部分
compressor = LLMChainExtractor.from_llm(ChatOpenAI(model="gpt-4"))
compression_retriever = ContextualCompressionRetriever(
    base_compressor=compressor,
    base_retriever=vectorstore.as_retriever(search_kwargs={"k": 5}),
)

# 使用自定义 Prompt 的 RAG Chain
qa_chain = RetrievalQA.from_chain_type(
    llm=ChatOpenAI(model="gpt-4", temperature=0),
    retriever=compression_retriever,
    chain_type_kwargs={"prompt": RAG_PROMPT},
)