核心场景​

智能客服、文档问答、代码生成、数据分析四大企业级应用落地。​

一、环境搭建与基础调用​

1. 依赖配置

# requirements.txt
langchain==0.1.10
openai==1.13.3
pinecone-client==3.2.0
python-dotenv==1.0.1
pandas==2.2.1

2. GPT-4.5 基础调用（含超时重试）

from langchain_openai import ChatOpenAI
from langchain_core.output_parsers import StrOutputParser
from tenacity import retry, stop_after_attempt, wait_exponential

# 初始化客户端
llm = ChatOpenAI(
    model="gpt-4.5-turbo",
    api_key="your-api-key",
    temperature=0.3,  # 降低随机性
    max_retries=3,
    timeout=30
)

# 带重试机制的生成函数
@retry(
    stop=stop_after_attempt(3),
    wait=wait_exponential(multiplier=1, min=2, max=10)
)
def generate_with_retry(prompt):
    chain = prompt | llm | StrOutputParser()
    return chain.invoke({"input": prompt})

# 示例：生成SQL查询
prompt = """根据用户问题生成PostgreSQL查询：
用户问题：查询2025年1月所有金额大于1000的订单，按创建时间倒序排列
表结构：orders(id BIGINT, amount DECIMAL(10,2), create_time TIMESTAMP, user_id BIGINT)
"""
print(generate_with_retry(prompt))

二、企业级文档问答系统​

1. 文档加载与向量存储

from langchain.document_loaders import PyPDFLoader
from langchain.text_splitter import RecursiveCharacterTextSplitter
from langchain.embeddings.openai import OpenAIEmbeddings
from langchain.vectorstores import Pinecone

# 加载PDF文档
loader = PyPDFLoader("企业产品手册.pdf")
documents = loader.load()

# 文本分割（避免token超限）
text_splitter = RecursiveCharacterTextSplitter(
    chunk_size=1000,
    chunk_overlap=200,
    length_function=len
)
chunks = text_splitter.split_documents(documents)

# 向量存储（Pinecone）
embeddings = OpenAIEmbeddings(model="text-embedding-3-small")
vector_store = Pinecone.from_documents(
    chunks,
    embeddings,
    index_name="product-manual-index"
)

2. 构建问答链（含上下文管理）

from langchain_core.prompts import ChatPromptTemplate
from langchain_core.runnables import RunnablePassthrough

# 定义提示词模板
prompt_template = ChatPromptTemplate.from_messages([
    ("system", "你是企业产品顾问，仅基于提供的文档回答问题，不要编造信息。"),
    ("human", "上下文：{context}\n问题：{question}")
])

# 检索器
retriever = vector_store.as_retriever(
    search_kwargs={"k": 3}  # 取Top3相关文档
)

# 构建链
chain = (
    {"context": retriever | (lambda docs: "\n".join([d.page_content for d in docs])), "question": RunnablePassthrough()}
    | prompt_template
    | llm
    | StrOutputParser()
)

# 问答示例
print(chain.invoke("产品X的最大并发支持多少？"))

三、智能客服 Agent 开发​

1. 工具定义与 Agent 配置

from langchain.agents import create_openai_tools_agent, AgentExecutor
from langchain_core.tools import Tool

# 定义工具：查询订单状态
def check_order_status(order_id: str) -> str:
    """查询订单状态的工具，输入订单号"""
    # 对接真实订单系统API
    response = requests.get(f"https://api.example.com/orders/{order_id}/status")
    return response.json()["status"]

# 工具列表
tools = [
    Tool(
        name="OrderStatusChecker",
        func=check_order_status,
        description="查询用户订单状态，需要传入订单号（格式：ORD2025xxxx）"
    )
]

# Agent提示词
agent_prompt = ChatPromptTemplate.from_messages([
    ("system", "你是智能客服，优先使用工具回答订单相关问题，无法回答时请引导用户联系人工。"),
    ("human", "{input}"),
    ("agent_scratchpad", "{agent_scratchpad}")
])

# 创建Agent
agent = create_openai_tools_agent(llm, tools, agent_prompt)
agent_executor = AgentExecutor(agent=agent, tools=tools, verbose=True)

# 运行Agent
agent_executor.invoke({
    "input": "我的订单ORD20251234为什么还没发货？"
})

四、私有化部署优化​

1. 模型缓存策略

# 使用Redis缓存频繁查询结果
import redis
from functools import lru_cache

redis_client = redis.Redis(host="localhost", port=6379, db=0)

def cache_with_redis(expire=3600):
    def decorator(func):
        @lru_cache(maxsize=1000)
        def wrapper(*args, **kwargs):
            key = f"llm_cache:{func.__name__}:{args}:{kwargs}"
            cached = redis_client.get(key)
            if cached:
                return cached.decode("utf-8")
            result = func(*args, **kwargs)
            redis_client.setex(key, expire, result)
            return result
        return wrapper
    return decorator

# 应用缓存
@cache_with_redis(expire=1800)
def generate_product_desc(product_name: str) -> str:
    prompt = f"生成{product_name}的产品描述（50字以内）"
    return generate_with_retry(prompt)

2. 成本控制技巧​

1. 非关键场景使用 gpt-3.5-turbo（成本仅为 GPT-4.5 的 1/10）​
2. 设置 max_tokens 上限（如 500），避免超长输出​
3. 批量处理请求，减少 API 调用次数​
4. 使用 Embedding 缓存，避免重复计算

如果您想要获取更多专业知识以及实战能力，欢迎订阅《程序员实战避坑手册：从面试到职场的问题一站式解决》专栏，专栏内容包含Java 后端开发实战避坑指南。适配各阶段 Java 开发者，直击开发全链路高频痛点，囊括 IDEA/Git 配置、Docker 环境搭建、MyBatis-Plus/SpringBoot 性能优化、MySQL 调优及大厂面试技巧。专栏内容均为实战案例 + 避坑步骤 + 落地解决方案，帮你规避开发陷阱。