LangChain 全景

简介

LangChain 是构建 LLM 应用的核心框架，提供链式调用、记忆管理、工具集成、向量检索等完整能力。

pip install langchain langchain-openai langchain-community

核心架构

LangChain 核心组件
├── Model I/O          # LLM 调用封装
│   ├── LLMs           # 文本补全模型
│   ├── ChatModels     # 对话模型
│   └── Prompts        # 提示词模板
├── Chains             # 链式调用
├── Memory             # 对话记忆
├── Agents             # 自主决策智能体
├── Tools              # 工具集成
├── Retrievers         # 向量检索
└── Callbacks          # 监控与日志

快速开始

from langchain_openai import ChatOpenAI
from langchain_core.messages import HumanMessage, SystemMessage

# 初始化模型（千问）
llm = ChatOpenAI(
    model="qwen-turbo",
    openai_api_key="sk-xxx",
    openai_api_base="https://dashscope.aliyuncs.com/compatible-mode/v1",
    temperature=0.7
)

# 直接调用
messages = [
    SystemMessage(content="你是专业的金融分析师"),
    HumanMessage(content="分析 A 股市场的主要风险")
]
response = llm.invoke(messages)
print(response.content)

Prompt Template

from langchain_core.prompts import ChatPromptTemplate, PromptTemplate

# 对话模板
chat_prompt = ChatPromptTemplate.from_messages([
    ("system", "你是一个专业的{role}，专注于{domain}领域"),
    ("human", "{question}")
])

# 格式化
messages = chat_prompt.format_messages(
    role="风险分析师",
    domain="信贷风控",
    question="小微企业贷款的主要风险有哪些？"
)

response = llm.invoke(messages)
print(response.content)

# Few-shot 模板
from langchain_core.prompts import FewShotChatMessagePromptTemplate

examples = [
    {
        "input": "这家公司资产负债率 85%",
        "output": "高风险：资产负债率超过 70% 警戒线，偿债压力较大"
    },
    {
        "input": "该企业连续三年营收增长 20%",
        "output": "低风险：持续稳定增长，经营状况良好"
    }
]

example_prompt = ChatPromptTemplate.from_messages([
    ("human", "{input}"),
    ("ai", "{output}")
])

few_shot_prompt = FewShotChatMessagePromptTemplate(
    example_prompt=example_prompt,
    examples=examples
)

final_prompt = ChatPromptTemplate.from_messages([
    ("system", "你是信贷风险评估专家，根据财务指标给出风险判断"),
    few_shot_prompt,
    ("human", "{input}")
])

LCEL（LangChain 表达式语言）

from langchain_core.output_parsers import StrOutputParser, JsonOutputParser
from langchain_core.prompts import ChatPromptTemplate

# 基础链：prompt | llm | parser
prompt = ChatPromptTemplate.from_template("用一句话总结：{text}")
parser = StrOutputParser()

chain = prompt | llm | parser

result = chain.invoke({"text": "招商银行2024年净利润同比增长5%，不良贷款率保持在0.95%"})
print(result)

# 并行链
from langchain_core.runnables import RunnableParallel

parallel_chain = RunnableParallel(
    summary=(ChatPromptTemplate.from_template("总结：{text}") | llm | StrOutputParser()),
    risks=(ChatPromptTemplate.from_template("列出风险点：{text}") | llm | StrOutputParser()),
)

results = parallel_chain.invoke({"text": "某银行年报内容..."})
print(results["summary"])
print(results["risks"])

输出解析器

from langchain_core.output_parsers import JsonOutputParser
from langchain_core.pydantic_v1 import BaseModel, Field
from typing import List

class RiskReport(BaseModel):
    company: str = Field(description="公司名称")
    risk_level: str = Field(description="风险等级：高/中/低")
    key_risks: List[str] = Field(description="主要风险点列表")
    score: float = Field(description="风险评分 0-100")

parser = JsonOutputParser(pydantic_object=RiskReport)

prompt = ChatPromptTemplate.from_messages([
    ("system", "你是信贷风险分析师，输出 JSON 格式的风险报告\n{format_instructions}"),
    ("human", "分析以下企业：{company_info}")
])

chain = prompt | llm | parser

result = chain.invoke({
    "company_info": "某科技公司，成立3年，营收500万，负债率60%",
    "format_instructions": parser.get_format_instructions()
})
print(f"风险等级: {result['risk_level']}")
print(f"风险评分: {result['score']}")

流式输出

# 同步流式
for chunk in chain.stream({"text": "详细分析..."}):
    print(chunk, end="", flush=True)

# 异步流式
import asyncio

async def async_stream():
    async for chunk in chain.astream({"text": "详细分析..."}):
        print(chunk, end="", flush=True)

asyncio.run(async_stream())

下一步

• LangChain Chains — 复杂链式调用
• LangChain Memory — 对话记忆管理
• LangChain Tools — 工具集成
• LangGraph — 有状态工作流