LangChain 四大核心模块详解

from langchain_openai import ChatOpenAI
from langchain_anthropic import ChatAnthropic
from langchain_community.chat_models import ChatLiteLLM

# OpenAI
openai_llm = ChatOpenAI(model="gpt-4o")

# Anthropic
claude_llm = ChatAnthropic(model="claude-4-opus")

# 统一接口调用
response = openai_llm.invoke("你好")
response = claude_llm.invoke("你好")  # 同样的调用方式

for chunk in llm.stream("写一首诗"):
    print(chunk.content, end="", flush=True)

responses = llm.batch([
    "你好",
    "今天天气怎么样",
    "帮我讲个笑话"
])

import asyncio

async def main():
    result = await llm.ainvoke("你好")
    return result

asyncio.run(main())

from langchain.prompts import PromptTemplate

template = PromptTemplate.from_template(
    "你是一个{role}，请用{style}风格回答以下问题：{question}"
)

prompt = template.format(
    role="数学老师",
    style="幽默",
    question="什么是微积分？"
)

from langchain.prompts import ChatPromptTemplate

template = ChatPromptTemplate.from_messages([
    ("system", "你是一个专业的{profession}"),
    ("human", "你好，{name}"),
    ("ai", "你好！有什么我可以帮助你的？"),
    ("human", "{question}")
])

prompt = template.format_messages(
    profession="律师",
    name="张三",
    question="合同纠纷怎么处理？"
)

from langchain.prompts import FewShotPromptTemplate
from langchain.prompts.example_selector import LengthBasedExampleSelector

examples = [
    {"input": "你好", "output": "你好！有什么可以帮你的？"},
    {"input": "再见", "output": "再见！有需要随时找我。"}
]

prompt = FewShotPromptTemplate(
    examples=examples,
    example_prompt=PromptTemplate.from_template(
        "输入: {input}\n输出: {output}"
    ),
    prefix="请根据以下示例回复：",
    suffix="输入: {input}\n输出:",
    input_variables=["input"]
)

from langchain.output_parsers import PydanticOutputParser
from pydantic import BaseModel

class Answer(BaseModel):
    sentiment: str
    confidence: float

parser = PydanticOutputParser(pydantic_object=Answer)

prompt = PromptTemplate(
    template="分析以下文本的情感：{text}\n{format_instructions}",
    input_variables=["text"],
    partial_variables={"format_instructions": parser.get_format_instructions()}
)

from langchain.chains import LLMChain

chain = LLMChain(
    llm=llm,
    prompt=prompt
)

result = chain.run(question="什么是AI？")

from langchain.chains import SimpleSequentialChain

chain1 = LLMChain(llm=llm, prompt=prompt1)
chain2 = LLMChain(llm=llm, prompt=prompt2)

overall_chain = SimpleSequentialChain(
    chains=[chain1, chain2],
    verbose=True
)

result = overall_chain.run("写一首关于春天的诗")

from langchain.chains import SequentialChain

chain1 = LLMChain(
    llm=llm,
    prompt=summary_prompt,
    output_key="summary"
)

chain2 = LLMChain(
    llm=llm,
    prompt=translate_prompt,
    input_variables=["summary"],
    output_key="translation"
)

overall_chain = SequentialChain(
    chains=[chain1, chain2],
    input_variables=["text"],
    output_variables=["summary", "translation"]
)

from langchain.chains import RouterChain
from langchain.chains.llm import LLMChain

# 定义子链
math_chain = LLMChain(llm=llm, prompt=math_prompt)
history_chain = LLMChain(llm=llm, prompt=history_prompt)

# 路由选择
router_chain = RouterChain(
    default_chain=default_chain,
    destination_chains={
        "math": math_chain,
        "history": history_chain
    }
)

from langchain.chains import LLMChain
from langchain.schema import BaseOutputParser

class UpperCaseParser(BaseOutputParser):
    def parse(self, text: str) -> str:
        return text.upper()

class CustomChain(LLMChain):
    def __init__(self, llm, prompt):
        super().__init__(llm=llm, prompt=prompt)
        self.output_parser = UpperCaseParser()
    
    def parse_output(self, output):
        return self.output_parser.parse(output)

from langchain.agents import AgentType, initialize_agent
from langchain.tools import Tool

tools = [
    Tool(
        name="搜索",
        func=search.run,
        description="用于搜索实时信息"
    ),
    Tool(
        name="计算",
        func=calculate.run,
        description="用于数学计算"
    )
]

agent = initialize_agent(
    tools,
    llm,
    agent=AgentType.ZERO_SHOT_REACT_DESCRIPTION,
    verbose=True
)

agent.run("搜索2024年诺贝尔物理学奖得主，并计算他们年龄之和")

from langgraph.prebuilt import create_react_agent
from langchain_openai import ChatOpenAI

llm = ChatOpenAI(model="gpt-4o")

# 定义工具
def get_weather(location: str) -> str:
    """获取指定位置的天气"""
    return f"{location} 今天是晴天，25度"

def calculate(expression: str) -> str:
    """计算数学表达式"""
    return str(eval(expression))

tools = [get_weather, calculate]

# 创建 Agent
agent = create_react_agent(llm, tools)

# 调用
result = agent.invoke({
    "messages": [("user", "北京天气怎么样？")]
})

from langchain_openai import ChatOpenAI
from langchain.prompts import ChatPromptTemplate
from langchain.chains import LLMChain
from langchain.agents import AgentType, initialize_agent
from langchain.tools import Tool

# 1. 初始化 LLM
llm = ChatOpenAI(model="gpt-4o")

# 2. 定义 Prompt 模板
prompt = ChatPromptTemplate.from_messages([
    ("system", "你是一个专业的{domain}助手"),
    ("human", "{question}")
])

# 3. 创建 Chain
chain = LLMChain(llm=llm, prompt=prompt)

# 4. 定义工具
def search_wiki(query):
    return f"关于{query}的搜索结果..."

tools = [
    Tool(
        name="搜索",
        func=search_wiki,
        description="搜索信息"
    )
]

# 5. 创建 Agent
agent = initialize_agent(
    tools,
    llm,
    agent=AgentType.ZERO_SHOT_REACT_DESCRIPTION
)

# 6. 执行
result = chain.run(domain="技术", question="什么是AI Agent？")

用户输入
    ↓
[Prompt 模板] → 添加上下文
    ↓
[Chain] → 按流程处理
    ↓
[Agent] → 决定是否使用工具
    ↓
[LLM] → 生成响应
    ↓
用户输出