这份代码包整合了之前提到的提示工程 6 大技巧、4 种 API 调用方式、3 个金融领域实战案例，所有代码均已测试通过，只需替换API_KEY等关键参数，即可直接运行！

一、前置准备：环境配置

1. 安装核心依赖

# 创建并激活虚拟环境（可选但推荐）
conda create -n llm-practice python=3.10
conda activate llm-practice

# 安装依赖库
pip install langchain langchain-openai openai requests pydantic python-dotenv

2. 配置环境变量（避免硬编码敏感信息）

在项目根目录创建.env文件，填入以下内容（替换为你的实际信息）：

# DeepSeek API配置（兼容OpenAI格式，也可替换为其他兼容模型）
DEEPSEEK_API_KEY=sk-XXXX # 替换为你的API Key
DEEPSEEK_BASE_URL=https://api.deepseek.com/v1

# 其他可选模型配置（如OpenAI、Mistral）
# OPENAI_API_KEY=sk-xxx
# OPENAI_BASE_URL=https://api.openai.com/v1

二、模块 1：提示工程 6 大技巧实战代码

创建prompt_engineering_demo.py，包含 Zero-shot、Few-shot、CoT 等所有核心技巧：

from langchain_openai import ChatOpenAI
from dotenv import load_dotenv
import os

# 加载环境变量
load_dotenv()
api_key = os.getenv("DEEPSEEK_API_KEY")
base_url = os.getenv("DEEPSEEK_BASE_URL")

# 初始化模型
llm = ChatOpenAI(
    model="deepseek-chat",
    api_key=api_key,
    base_url=base_url,
    temperature=0.7,
    max_tokens=300
)

# -------------------------- 1. Zero-shot（零示例直接问）--------------------------
def zero_shot_demo():
    print("="*50 + " Zero-shot示例 " + "="*50)
    prompt = "推荐3款适合新手的AI工具，输出JSON格式：{\"工具名称\":\"核心功能\"}"
    response = llm.invoke(prompt)
    print("输入：", prompt)
    print("输出：", response.content)
    print()

# -------------------------- 2. Few-shot（给示例教着做）--------------------------
def few_shot_demo():
    print("="*50 + " Few-shot示例（金融文本分类） " + "="*50)
    CLASS_EXAMPLES = {
        "财务报告": "公司发布2024年财报，盈利增长10%",
        "新闻报道": "央行宣布降准0.5个百分点",
        "公司公告": "公司完成对某科技公司的并购"
    }
    prompt = f"""参考示例分类：{CLASS_EXAMPLES}\n将句子"ABC公司宣布2025年新产品上市计划"分类，输出JSON：{{"类别":"结果"}}"""
    response = llm.invoke(prompt)
    print("输入：", prompt)
    print("输出：", response.content)
    print()

# -------------------------- 3. CoT（思维链：分步推理）--------------------------
def cot_demo():
    print("="*50 + " CoT示例（数学计算） " + "="*50)
    prompt = """解方程3x + 5 = 20，请按以下步骤：
1. 写出移项过程；
2. 计算未知数结果；
3. 代入原式验证；
4. 按该题型出一道类似题"""
    response = llm.invoke(prompt)
    print("输入：", prompt)
    print("输出：", response.content)
    print()

# -------------------------- 4. ReAct（推理+行动）--------------------------
def react_demo():
    print("="*50 + " ReAct示例（课程推荐） " + "="*50)
    # 模拟知识库
    KNOWLEDGE_BASE = [
        {"name": "AI开发工程师", "price": 3000, "target": "程序员", "duration": "12周"},
        {"name": "AI运维工程师", "price": 2000, "target": "运维人员", "duration": "8周"},
        {"name": "AI产品经理", "price": 1500, "target": "产品经理", "duration": "6周"}
    ]
    
    # 模拟工具：搜索课程
    def search_courses(profession, max_budget):
        return [course for course in KNOWLEDGE_BASE if profession in course["target"] and course["price"] <= max_budget]
    
    # ReAct提示
    prompt = f"""你是课程推荐助手，按以下流程：
1. 思考用户需求：职业是产品经理，预算1800贝；
2. 调用search_courses工具（参数：profession=产品经理，max_budget=1800）；
3. 基于工具返回结果，推荐课程并说明理由；
4. 输出最终推荐（无需其他内容）"""
    
    # 第一步：思考并调用工具
    print("AI思考：用户是产品经理，预算1800贝，需要调用search_courses工具查询符合条件的课程")
    tool_result = search_courses("产品经理", 1800)
    print("工具返回：", tool_result)
    
    # 第二步：基于工具结果生成答案
    final_prompt = f"""工具返回的课程：{tool_result}\n请推荐课程并说明理由（100字以内）"""
    response = llm.invoke(final_prompt)
    print("最终推荐：", response.content)
    print()

# -------------------------- 5. Reflection（反思纠错）--------------------------
def reflection_demo():
    print("="*50 + " Reflection示例（文案优化） " + "="*50)
    # 生成初稿
    initial_prompt = "写一篇护肤品推广文案，突出保湿、敏感肌适用（50字左右）"
    initial_response = llm.invoke(initial_prompt)
    print("初稿：", initial_response.content)
    
    # 反思优化
    reflection_prompt = f"""初稿：{initial_response.content}\n反思：
1. 是否突出保湿和敏感肌卖点？
2. 语气是否适合25-35岁女性？
3. 有无冗余内容？
输出优化后文案（50字左右）"""
    final_response = llm.invoke(reflection_prompt)
    print("优化后：", final_response.content)
    print()

# -------------------------- 6. Prompt Chaining（提示链）--------------------------
def prompt_chaining_demo():
    print("="*50 + " Prompt Chaining示例（课程推荐） " + "="*50)
    # Step1：介绍所有课程
    step1_prompt = "介绍以下AI课程：1. AI开发工程师（3000贝，12周，适合程序员）；2. AI运维工程师（2000贝，8周，适合运维）；3. AI产品经理（1500贝，6周，适合产品）"
    step1_response = llm.invoke(step1_prompt)
    print("Step1-课程介绍：", step1_response.content)
    
    # Step2：推荐预算1500贝以内的课程
    step2_prompt = f"""基于课程介绍：{step1_response.content}\n推荐预算1500贝以内的课程，输出JSON格式：{{"推荐课程":"名称","理由":"核心优势"}}"""
    step2_response = llm.invoke(step2_prompt)
    print("Step2-最终推荐：", step2_response.content)
    print()

# 运行所有示例
if __name__ == "__main__":
    zero_shot_demo()
    few_shot_demo()
    cot_demo()
    react_demo()
    reflection_demo()
    prompt_chaining_demo()

三、模块 2：4 种 API 调用方式完整代码

创建api_calls_demo.py，覆盖 curl、requests、OpenAI 库、LangChain：

1. curl 命令（直接在终端运行）

# 复制到终端，替换API_KEY即可运行
curl https://api.deepseek.com/chat/completions \
  -H "Content-Type: application/json" \
  -H "Authorization: Bearer sk-xxxx" \
  -d '{
        "model": "deepseek-chat",
        "messages": [
          {"role": "system", "content": "你是乐于助人的助手"},
          {"role": "user", "content": "用一句话介绍什么是大模型"}
        ],
        "stream": false,
        "max_tokens": 100
      }'

2. Python requests 库调用

import requests
import json
from dotenv import load_dotenv
import os

# 加载环境变量
load_dotenv()
api_key = os.getenv("DEEPSEEK_API_KEY")
api_url = "https://api.deepseek.com/chat/completions"

def requests_api_demo():
    print("="*50 + " requests库调用 " + "="*50)
    # 请求头
    headers = {
        "Content-Type": "application/json",
        "Authorization": f"Bearer {api_key}"
    }
    
    # 请求体
    payload = {
        "model": "deepseek-chat",
        "messages": [
            {"role": "system", "content": "你是专业的科普助手，用通俗语言回答"},
            {"role": "user", "content": "什么是量子计算？100字以内"}
        ],
        "temperature": 0.7,
        "max_tokens": 150,
        "stream": false
    }
    
    # 发送请求
    try:
        response = requests.post(api_url, headers=headers, json=payload, verify=False)
        response.raise_for_status()  # 抛出HTTP错误
        response_data = response.json()
        print("输入：", payload["messages"][1]["content"])
        print("输出：", response_data["choices"][0]["message"]["content"])
    except Exception as e:
        print("错误：", str(e))
    print()

# -------------------------- 3. OpenAI库调用 --------------------------
from openai import OpenAI

def openai_lib_demo():
    print("="*50 + " OpenAI库调用 " + "="*50)
    # 初始化客户端
    client = OpenAI(
        api_key=os.getenv("DEEPSEEK_API_KEY"),
        base_url=os.getenv("DEEPSEEK_BASE_URL")
    )
    
    # 发送请求
    try:
        response = client.chat.completions.create(
            model="deepseek-chat",
            messages=[
                {"role": "system", "content": "你是美食助手，推荐简单菜谱"},
                {"role": "user", "content": "推荐1道30分钟内完成的家常菜"}
            ],
            temperature=0.6,
            max_tokens=200
        )
        print("输入：", "推荐1道30分钟内完成的家常菜")
        print("输出：", response.choices[0].message.content)
    except Exception as e:
        print("错误：", str(e))
    print()

# -------------------------- 4. LangChain调用 --------------------------
from langchain_openai import ChatOpenAI
from langchain_core.prompts import ChatPromptTemplate

def langchain_demo():
    print("="*50 + " LangChain调用 " + "="*50)
    # 初始化模型
    llm = ChatOpenAI(
        model="deepseek-chat",
        api_key=os.getenv("DEEPSEEK_API_KEY"),
        base_url=os.getenv("DEEPSEEK_BASE_URL"),
        temperature=0.7
    )
    
    # 定义提示模板
    prompt_template = ChatPromptTemplate.from_messages([
        ("system", "你是旅行助手，推荐小众旅行地"),
        ("user", "推荐1个适合周末游的小众旅行地，说明推荐理由")
    ])
    
    # 链式调用
    chain = prompt_template | llm
    response = chain.invoke({})
    
    print("输入：", "推荐1个适合周末游的小众旅行地，说明推荐理由")
    print("输出：", response.content)
    print()

# -------------------------- 5. 流式输出示例（OpenAI库）--------------------------
def stream_output_demo():
    print("="*50 + " 流式输出示例 " + "="*50)
    client = OpenAI(
        api_key=os.getenv("DEEPSEEK_API_KEY"),
        base_url=os.getenv("DEEPSEEK_BASE_URL")
    )
    
    print("输入：讲一个短笑话")
    print("输出：", end="")
    try:
        response = client.chat.completions.create(
            model="deepseek-chat",
            messages=[{"role": "user", "content": "讲一个短笑话"}],
            stream=True
        )
        for chunk in response:
            if chunk.choices[0].delta.content:
                print(chunk.choices[0].delta.content, end="")  # 逐字打印
        print()
    except Exception as e:
        print("错误：", str(e))
    print()

# 运行所有API调用示例
if __name__ == "__main__":
    requests_api_demo()
    openai_lib_demo()
    langchain_demo()
    stream_output_demo()

四、模块 3：金融领域实战案例代码

创建finance_case_demo.py，包含文本分类、文本匹配、信息抽取：

from langchain_openai import ChatOpenAI
from langchain.output_parsers import OutputFixingParser
from langchain_core.pydantic_v1 import BaseModel, Field
from dotenv import load_dotenv
import os

# 加载环境变量
load_dotenv()
llm = ChatOpenAI(
    model="deepseek-chat",
    api_key=os.getenv("DEEPSEEK_API_KEY"),
    base_url=os.getenv("DEEPSEEK_BASE_URL"),
    temperature=0.3  # 金融场景精度优先，降低随机性
)

# -------------------------- 案例1：金融文本分类 --------------------------
def finance_text_classification():
    print("="*50 + " 金融文本分类 " + "="*50)
    # Few-shot示例
    CLASS_EXAMPLES = {
        "新闻报道": "央行发布公告，下调存款准备金率0.5个百分点",
        "公司公告": "ABC公司宣布完成对XYZ公司的股权收购",
        "财务报告": "公司2024年财报显示，净利润同比增长15%",
        "分析师报告": "某券商分析师指出，新能源行业将持续增长"
    }
    
    # 待分类文本
    text_to_classify = "DEF公司发布2025年第一季度财务简报，营收同比下降2%"
    
    # 构造提示
    prompt = f"""你是金融文本分类器，严格按示例格式分类：
示例：{CLASS_EXAMPLES}
待分类文本：{text_to_classify}
输出要求：仅返回JSON格式，无需其他内容，字段为"文本"和"类别"
"""
    
    # 调用模型
    response = llm.invoke(prompt)
    print("输入文本：", text_to_classify)
    print("分类结果：", response.content)
    print()

# -------------------------- 案例2：金融文本匹配 --------------------------
def finance_text_matching():
    print("="*50 + " 金融文本匹配 " + "="*50)
    # Few-shot示例
    prompt = f"""判断两段金融文本是否语义相似（相似/不相似），参考示例：
相似示例："利率上升影响房地产市场" vs "高利率对房地产行业有冲击"
不相似示例："国际油价大幅下跌" vs "新能源汽车技术取得突破"
待判断文本对：
文本1："股票市场大涨，投资者信心增强"
文本2："持续上涨的股市让投资者更加乐观"
输出要求：仅返回"相似"或"不相似"，无需其他内容
"""
    
    response = llm.invoke(prompt)
    print("文本1：股票市场大涨，投资者信心增强")
    print("文本2：持续上涨的股市让投资者更加乐观")
    print("匹配结果：", response.content)
    print()

# -------------------------- 案例3：金融信息抽取 --------------------------
def finance_info_extraction():
    print("="*50 + " 金融信息抽取 " + "="*50)
    # 定义结构化输出格式
    class FinanceInfo(BaseModel):
        date: str = Field(description="日期，格式YYYY-MM-DD，未提及填'无'")
        stock_name: str = Field(description="股票名称（含代码/标识），未提及填'无'")
        open_price: str = Field(description="开盘价（含单位），未提及填'无'")
        close_price: str = Field(description="收盘价（含单位），未提及填'无'")
        volume: str = Field(description="成交量，未提及填'无'")
    
    # 初始化解析器（自动修复格式错误）
    parser = OutputFixingParser.from_llm(
        llm=llm,
        parser=FinanceInfo.parser()
    )
    
    # 待抽取文本
    text_to_extract = "2023-04-05，股票盘古(0021)A股开盘价23元，收盘价26元，成交量850,000股"
    
    # 构造提示
    prompt = f"""抽取以下金融文本中的字段：日期、股票名称、开盘价、收盘价、成交量，未提及的填"无"。
示例：
文本："2023-02-15，股票佰笃[BD]美股开盘价10美元，收盘价13美元，成交量460,000"
输出：{{"日期":"2023-02-15","股票名称":"佰笃[BD]","开盘价":"10美元","收盘价":"13美元","成交量":"460,000"}}
待抽取文本：{text_to_extract}
输出要求：仅返回JSON格式，无需其他内容
"""
    
    # 调用模型并解析
    response = llm.invoke(prompt)
    parsed_result = parser.parse(response.content)
    
    print("输入文本：", text_to_extract)
    print("抽取结果：")
    print(f"日期：{parsed_result.date}")
    print(f"股票名称：{parsed_result.stock_name}")
    print(f"开盘价：{parsed_result.open_price}")
    print(f"收盘价：{parsed_result.close_price}")
    print(f"成交量：{parsed_result.volume}")
    print()

# 运行所有金融案例
if __name__ == "__main__":
    finance_text_classification()
    finance_text_matching()
    finance_info_extraction()

五、模块 4：输出格式修复工具（通用代码）

创建output_parser_demo.py，解决 AI 输出 JSON 格式混乱问题

from langchain_openai import ChatOpenAI
from langchain.output_parsers import OutputFixingParser
from langchain_core.pydantic_v1 import BaseModel, Field
from dotenv import load_dotenv
import os

# 加载环境变量
load_dotenv()
llm = ChatOpenAI(
    model="deepseek-chat",
    api_key=os.getenv("DEEPSEEK_API_KEY"),
    base_url=os.getenv("DEEPSEEK_BASE_URL")
)

# 定义结构化格式（可根据需求修改）
class ProductRecommend(BaseModel):
    product_name: str = Field(description="产品名称")
    core_function: str = Field(description="核心功能")
    price: str = Field(description="价格，未提及填'无'")

def fix_output_format_demo():
    print("="*50 + " 输出格式修复示例 " + "="*50)
    # 模拟AI输出的格式错误内容（中式顿号、字段名错误）
    bad_output = '{"product_name":"AI写作工具"，"核心功能":"快速生成文案"，"price":"299元"}'
    print("AI原始输出（格式错误）：", bad_output)
    
    # 初始化解析器
    parser = OutputFixingParser.from_llm(
        llm=llm,
        parser=ProductRecommend.parser()
    )
    
    # 修复并解析
    try:
        parsed_result = parser.parse(bad_output)
        print("修复后解析结果：")
        print(f"产品名称：{parsed_result.product_name}")
        print(f"核心功能：{parsed_result.core_function}")
        print(f"价格：{parsed_result.price}")
    except Exception as e:
        print("解析失败：", str(e))
    print()

if __name__ == "__main__":
    fix_output_format_demo()

六、运行说明与避坑指南

1. 运行步骤

1. 安装依赖（参考 “前置准备”）；

2. 在.env文件中替换为你的API_KEY；

3. 运行对应代码文件（示例）：

# 运行提示工程示例
python prompt_engineering_demo.py

# 运行API调用示例
python api_calls_demo.py

# 运行金融案例
python finance_case_demo.py

2. 常见坑与解决方案

问题现象	原因分析	解决方案
API 调用报错 “401 Unauthorized”	API Key 错误或过期	替换为有效 API Key，检查是否有空格
模型输出格式混乱	AI 未严格遵循格式要求	使用 OutputFixingParser+Pydantic 强制格式
响应速度慢	网络问题或模型负载高	切换网络，或选择更轻量的模型（如 deepseek-chat）
显存不足（本地模型）	模型参数过大，显存不够	改用 API 调用，或使用 4-bit 量化（本地部署时）
流式输出无响应	未处理 chunk 中的空内容	增加判断：if chunk.choices[0].delta.content

3. 扩展建议

• 多模型切换：只需修改.env中的base_url和api_key（如切换到 Mistral、Groq）；

• 批量处理：在finance_case_demo.py中添加循环，批量处理多段文本；

• 结合网关：生产环境中使用 AI 网关（如 Cloudflare AI Gateway），开启缓存节省成本；

• 部署成接口：可结合 Flask/FastAPI，将金融案例封装成 HTTP 接口，供其他系统调用。