第1章: LangChain 入门：用最短路径搭起 LLM 应用骨架

from langchain_community.llms import Tongyi
import os

# 设置API Key（从环境变量获取）
# 从环境变量读取 API Key（请确保已设置 DASHSCOPE_API_KEY）
os.environ["DASHSCOPE_API_KEY"] = os.environ.get("DASHSCOPE_API_KEY", "")

# 初始化通义千问模型
llm = Tongyi(
    model_name="qwen-turbo",  # 模型名称
    temperature=0.7,           # 温度参数
    top_p=0.8,                # top_p采样
    max_tokens=2000           # 最大生成长度
)

# 直接调用
response = llm.invoke("介绍一下Python语言")
print(response)

from langchain_community.llms import Tongyi
import os

# 从环境变量读取 API Key（请确保已设置 DASHSCOPE_API_KEY）
os.environ["DASHSCOPE_API_KEY"] = os.environ.get("DASHSCOPE_API_KEY", "")

# 开启 streaming=True
llm = Tongyi(model_name="qwen-turbo", temperature=0.7, streaming=True)

# 逐段输出（像打字一样）
for chunk in llm.stream("用通俗的话解释一下：什么是LangChain？"):
    print(chunk, end="", flush=True)
print()

"""
Token计数示例 - 了解文本消耗多少Token
"""
import tiktoken

# 获取tokenizer（使用cl100k_base作为近似估算）
encoding = tiktoken.get_encoding("cl100k_base")

# ========== 示例1：中文文本 ==========
text_cn = "黄河之水天上来"
tokens_cn = encoding.encode(text_cn)
print(f"中文文本: {text_cn}")
print(f"Token编码: {tokens_cn}")
print(f"Token数量: {len(tokens_cn)}")
print()

# ========== 示例2：英文文本 ==========
text_en = "Hello, how are you?"
tokens_en = encoding.encode(text_en)
print(f"英文文本: {text_en}")
print(f"Token编码: {tokens_en}")
print(f"Token数量: {len(tokens_en)}")
print()

# ========== 示例3：混合文本 ==========
text_mix = "我喜欢Python编程，it's amazing!"
tokens_mix = encoding.encode(text_mix)
print(f"混合文本: {text_mix}")
print(f"Token数量: {len(tokens_mix)}")
print()

# ========== 示例4：计算对话的Token数 ==========
messages = [
    {"role": "system", "content": "你是一个Python助手"},
    {"role": "user", "content": "什么是列表？"},
    {"role": "assistant", "content": "列表是Python中的一种数据结构..."}
]

total_tokens = 0
for msg in messages:
    content = msg["content"]
    tokens = encoding.encode(content)
    total_tokens += len(tokens)
    print(f"{msg['role']}: {len(tokens)} tokens")

print(f"\n总Token数: {total_tokens}")

from langchain_community.chat_models import ChatTongyi
from langchain_core.messages import HumanMessage
import os

# 从环境变量读取 API Key（请确保已设置 DASHSCOPE_API_KEY）
os.environ["DASHSCOPE_API_KEY"] = os.environ.get("DASHSCOPE_API_KEY", "")

chat_model = ChatTongyi(model_name="qwen-turbo", temperature=0)

response = chat_model.invoke([HumanMessage(content="介绍一下Python语言")])
print(f"回复: {response.content}")

# 尝试从返回消息中读取 token usage（不同版本/不同模型字段名可能略有差异）
usage = None
for key in ("usage_metadata", "response_metadata", "token_usage"):
    value = getattr(response, key, None)
    if value:
        usage = value
        break

print(f"\n--- Token统计（从AIMessage元数据读取） ---")
print(usage)

from langchain_community.llms import Tongyi
import os

# 从环境变量读取 API Key（请确保已设置 DASHSCOPE_API_KEY）
os.environ["DASHSCOPE_API_KEY"] = os.environ.get("DASHSCOPE_API_KEY", "")

# 测试问题
question = "用一句话描述春天"

# 实验1：低温度（精确、确定）
print("=== Temperature = 0.1 ===")
llm_low = Tongyi(model_name="qwen-turbo", temperature=0.1)
for i in range(3):
    print(f"第{i+1}次: {llm_low.invoke(question)}")

# 实验2：中温度（平衡）
print("\n=== Temperature = 0.7 ===")
llm_mid = Tongyi(model_name="qwen-turbo", temperature=0.7)
for i in range(3):
    print(f"第{i+1}次: {llm_mid.invoke(question)}")

# 实验3：高温度（创意、随机）
print("\n=== Temperature = 1.5 ===")
llm_high = Tongyi(model_name="qwen-turbo", temperature=1.5)
for i in range(3):
    print(f"第{i+1}次: {llm_high.invoke(question)}")

from langchain_community.llms import Tongyi
from langchain_community.chat_models import ChatTongyi

# 文本生成模型（适合单次生成）
llm = Tongyi(model_name="qwen-turbo")
response = llm.invoke("什么是Python？")

# 对话模型（支持多轮对话）
chat_model = ChatTongyi(model_name="qwen-turbo")
messages = [
    ("system", "你是Python专家"),
    ("user", "什么是列表？")
]
response = chat_model.invoke(messages)
print(response.content)

from langchain_community.chat_models import ChatTongyi
from langchain_core.messages import SystemMessage, HumanMessage, AIMessage

chat_model = ChatTongyi(model_name="qwen-turbo")

messages = [
    SystemMessage(content="你是Python老师，用通俗中文回答，尽量给出例子"),
    HumanMessage(content="解释一下什么是列表（list）"),
]

result = chat_model.invoke(messages)
print(type(result))          # <class 'langchain_core.messages.ai.AIMessage'>
print(result.content)        # 模型输出文本

# 如果你要把历史对话继续带上去，也可以显式追加 AIMessage
messages.append(AIMessage(content=result.content))
messages.append(HumanMessage(content="再给一个包含 for 循环的例子"))
result2 = chat_model.invoke(messages)
print(result2.content)

from langchain_core.prompts import PromptTemplate, ChatPromptTemplate

# 简单提示模板
prompt = PromptTemplate.from_template("将{text}翻译成{language}")
formatted = prompt.format(text="你好", language="英文")
print(formatted)  # 输出：将你好翻译成英文

# 对话提示模板
chat_prompt = ChatPromptTemplate.from_messages([
    ("system", "你是{role}"),
    ("user", "{input}")
])
messages = chat_prompt.format_messages(
    role="Python老师",
    input="解释什么是变量"
)
print(messages)

from langchain_core.output_parsers import StrOutputParser, JsonOutputParser, CommaSeparatedListOutputParser
from langchain_community.llms import Tongyi
import os

llm = Tongyi(model_name="qwen-turbo")

# ========== 1. StrOutputParser - 字符串解析器 ==========
# 最常用！将输出转换为纯字符串
str_parser = StrOutputParser()
chain = llm | str_parser

result = chain.invoke("用一句话介绍Python")
print(f"类型: {type(result)}")  # 
print(f"内容: {result}")


# ========== 2. JsonOutputParser - JSON解析器 ==========
# 将输出解析为JSON对象
from langchain_core.prompts import PromptTemplate

json_parser = JsonOutputParser()
prompt = PromptTemplate.from_template(
    """请将以下用户信息提取为JSON格式：\n{user_input}\n\n{format_instructions}"""
)

chain = prompt | llm | json_parser
result = chain.invoke({
    "user_input": "介绍一下李明，30岁，程序员",
    "format_instructions": json_parser.get_format_instructions()
})
print(f"类型: {type(result)}")  # 
print(f"内容: {result}")  # {'姓名': '李明', '年龄': 30, '职业': '程序员'}


# ========== 3. CommaSeparatedListOutputParser - 逗号分隔列表解析器 ==========
# 将输出解析为列表
list_parser = CommaSeparatedListOutputParser()
prompt = PromptTemplate.from_template(
    """请列出3种编程语言，用逗号分隔：\n{format_instructions}"""
)

chain = prompt | llm | list_parser
result = chain.invoke({"format_instructions": list_parser.get_format_instructions()})
print(f"类型: {type(result)}")  # 
print(f"内容: {result}")  # ['Python', 'Java', 'JavaScript']


# ========== 4. PydanticOutputParser - Pydantic结构化解析器 ==========
# 把输出解析为“强类型对象”（字段缺失/类型不对会报错）
from pydantic import BaseModel, Field
from langchain_core.output_parsers import PydanticOutputParser


class UserInfo(BaseModel):
    name: str = Field(description="姓名")
    age: int = Field(description="年龄")
    job: str = Field(description="职业")


pydantic_parser = PydanticOutputParser(pydantic_object=UserInfo)
pyd_prompt = PromptTemplate.from_template(
    """从文本中提取用户信息，并按指定 JSON 格式输出：\n{text}\n\n{format_instructions}"""
)

pyd_chain = pyd_prompt | llm | pydantic_parser
pyd_result = pyd_chain.invoke({
    "text": "李明，30岁，程序员",
    "format_instructions": pydantic_parser.get_format_instructions(),
})
print(f"类型: {type(pyd_result)}")
print(f"内容: {pyd_result}")


# ========== 5. OutputFixingParser - 解析失败自动修正（真实使用） ==========
# 真实使用 OutputFixingParser：当解析失败时，自动让模型修正后重试
from langchain_classic.output_parsers import OutputFixingParser

# 创建一个基础 parser（这里用 JsonOutputParser）
base_parser = JsonOutputParser()

# 创建 OutputFixingParser，传入 LLM 和基础 parser
fixing_parser = OutputFixingParser.from_llm(
    llm=llm,
    parser=base_parser
)

# 测试：故意给一个错误的 JSON
bad_json = "{'name': 'Li Ming', 'age': 30, 'job': 'Engineer'}"  # 单引号，不是合法 JSON
try:
    fixed_demo = fixing_parser.parse(bad_json)
    print(f"✅ 修复成功！类型: {type(fixed_demo)}")
    print(f"内容: {fixed_demo}")
except Exception as e:
    print(f"❌ 修复失败: {e}")


# ========== 6. RetryWithErrorOutputParser - 带错误原因的重试（真实使用） ==========
# 真实使用 RetryWithErrorOutputParser：解析失败时把错误原因喂回模型重试
from langchain_classic.output_parsers import RetryWithErrorOutputParser

# 创建 RetryWithErrorOutputParser
retry_parser = RetryWithErrorOutputParser.from_llm(
    llm=llm,
    parser=base_parser
)

# 测试：给一个完全不是 JSON 的格式
raw_model_output = "姓名=李明; 年龄=三十; 职业=程序员"  # 非 JSON
try:
    retry_demo = retry_parser.parse(raw_model_output)
    print(f"✅ 重试成功！类型: {type(retry_demo)}")
    print(f"内容: {retry_demo}")
except Exception as e:
    print(f"❌ 重试失败: {e}")

from langchain_community.llms import Tongyi
from langchain_core.prompts import PromptTemplate
from langchain_core.output_parsers import StrOutputParser
import os

# 从环境变量读取 API Key（请确保已设置 DASHSCOPE_API_KEY）
os.environ["DASHSCOPE_API_KEY"] = os.environ.get("DASHSCOPE_API_KEY", "")
llm = Tongyi(model_name="qwen-turbo")
prompt = PromptTemplate.from_template("用一句话介绍{topic}")

# ========== 方式对比：传统 vs LCEL ==========

# 传统方式（手写 prompt + 调用模型）
text = prompt.format(topic="人工智能")
result = llm.invoke(text)
print(result)

# LCEL方式（推荐）⭐
chain_new = prompt | llm | StrOutputParser()
result = chain_new.invoke({"topic": "人工智能"})
print(result)  # 直接返回字符串，更简洁！


# ========== LCEL完整示例 ==========
# 创建链：prompt | model | output_parser
chain = prompt | llm | StrOutputParser()

# 方法1：invoke() - 单次调用
result = chain.invoke({"topic": "人工智能"})
print(result)
# 输出：人工智能是模拟人类智能的技术...


# ========== 方法2：batch() - 批量处理 ==========
# 一次处理多个输入
results = chain.batch([
    {"topic": "人工智能"},
    {"topic": "区块链"},
    {"topic": "量子计算"}
])
for i, result in enumerate(results, 1):
    print(f"{i}. {result}")


# ========== 方法3：stream() - 流式输出 ==========
# 像打字一样逐字显示
for chunk in chain.stream({"topic": "机器学习"}):
    print(chunk, end="", flush=True)


# ========== LCEL顺序链示例 ==========
# 使用 | 操作符串联多个步骤
prompt1 = PromptTemplate.from_template("为{topic}写一个故事大纲")
prompt2 = PromptTemplate.from_template("根据以下大纲，写一个详细的故事：\n{outline}")

# 方式1：分步创建
outline_chain = prompt1 | llm | StrOutputParser()
story_chain = prompt2 | llm | StrOutputParser()

# 先生成大纲
outline = outline_chain.invoke({"topic": "太空探险"})
print(f"大纲：{outline}")

# 再生成故事
story = story_chain.invoke({"outline": outline})
print(f"故事：{story}")


# 方式2：使用RunnablePassthrough自动传递
from langchain_core.runnables import RunnablePassthrough

# 完整的链式调用
full_chain = (
    {"outline": prompt1 | llm | StrOutputParser()}  # 生成大纲
    | prompt2  # 使用大纲生成故事
    | llm
    | StrOutputParser()
)
story = full_chain.invoke({"topic": "太空探险"})
print(story)

from langchain_community.llms import Tongyi
from langchain_core.prompts import PromptTemplate
from langchain_core.output_parsers import StrOutputParser
from langchain_core.runnables import RunnableBranch, RunnableLambda
import os

# 从环境变量读取 API Key（请确保已设置 DASHSCOPE_API_KEY）
os.environ["DASHSCOPE_API_KEY"] = os.environ.get("DASHSCOPE_API_KEY", "")
llm = Tongyi(model_name="qwen-turbo", temperature=0)
parser = StrOutputParser()

# 1) Router：识别意图，只输出一个标签：translate / summarize / explain
router_prompt = PromptTemplate.from_template(
"""你是一个路由器，只输出一个标签：translate / summarize / explain。

用户输入：{text}

判断规则：
- 如果用户要求翻译（含“翻译/英文/中文/日文”等），输出 translate
- 如果用户要求总结/摘要/要点，输出 summarize
- 如果用户给了代码或要求解释代码，输出 explain

只输出标签，不要输出其他内容。"""
)

router_chain = router_prompt | llm | parser


# 2) 三个子链
translate_prompt = PromptTemplate.from_template("把下面内容翻译成英文：\n{text}\n\n英文：")
summarize_prompt = PromptTemplate.from_template("把下面内容总结成 3 条要点：\n{text}\n\n要点：")
explain_prompt = PromptTemplate.from_template(
"""你是资深开发者。请解释下面代码的作用、关键步骤，并指出一个可改进点：

{text}
"""
)

translate_chain = translate_prompt | llm | parser
summarize_chain = summarize_prompt | llm | parser
explain_chain = explain_prompt | llm | parser


# 3) 路由 + 分支执行
def pick_intent(x: dict) -> str:
    intent = router_chain.invoke({"text": x["text"]}).strip().lower()
    return intent

intent_runnable = RunnableLambda(pick_intent)

routed_chain = (
    {"intent": intent_runnable, "text": RunnableLambda(lambda x: x["text"])}
    | RunnableBranch(
        (lambda x: x["intent"] == "translate", translate_chain),
        (lambda x: x["intent"] == "summarize", summarize_chain),
        (lambda x: x["intent"] == "explain", explain_chain),
        explain_chain,
    )
)


print("=== 1) 翻译 ===")
print(routed_chain.invoke({"text": "把这句话翻译成英文：我正在学习 LangChain"}))

print("\n=== 2) 总结 ===")
print(routed_chain.invoke({"text": "请把下面这段话总结一下：LangChain 是一个用来构建大模型应用的框架，它提供了提示词、链、记忆、检索等组件。"}))

print("\n=== 3) 解释代码 ===")
print(routed_chain.invoke({"text": "def add(a, b):\n    return a + b"}))

from langchain_community.llms import Tongyi
from langchain_core.prompts import PromptTemplate
from langchain_core.output_parsers import StrOutputParser
import os

# 初始化模型（翻译任务用低温度）
llm = Tongyi(
    model_name="qwen-turbo",
    temperature=0.1,  # 低温度，保证翻译准确
    top_p=0.8
)

# 创建提示模板
template = """你是一个专业的翻译助手。
请将以下{source_lang}文本翻译成{target_lang}，要求：
1. 准确传达原文含义
2. 符合目标语言习惯
3. 保持专业术语的准确性

原文：{text}

翻译："""

prompt = PromptTemplate(
    input_variables=["source_lang", "target_lang", "text"],
    template=template
)

# 使用新的LCEL语法创建链（替代已弃用的LLMChain）
translation_chain = prompt | llm | StrOutputParser()

# 使用示例
result = translation_chain.invoke({
    "source_lang": "中文",
    "target_lang": "英文",
    "text": "人工智能正在改变我们的生活方式"
})

print(result)

from langchain_community.llms import Tongyi
from langchain_core.prompts import PromptTemplate
from langchain_core.output_parsers import StrOutputParser
import os
from dotenv import load_dotenv

# 加载.env文件中的环境变量
load_dotenv()

# 检查API密钥是否已设置
api_key = os.getenv("DASHSCOPE_API_KEY")
if not api_key or api_key == "your-actual-api-key":
    print("错误：请设置有效的DASHSCOPE_API_KEY环境变量")
    print("您可以在.env文件中添加：DASHSCOPE_API_KEY=your-actual-api-key")
    exit(1)

# 设置API Key
os.environ["DASHSCOPE_API_KEY"] = api_key

# 初始化模型
llm = Tongyi(
    model_name="qwen-turbo",
    temperature=0.3,  # 中低温度，保证解释准确
    max_tokens=1000
)

# 创建提示模板
template = """你是一个编程专家。请详细解释以下{language}代码的功能：

代码：
{code}

请从以下几个方面解释：
1. 代码的主要功能
2. 关键步骤的作用
3. 可能的应用场景

解释："""

prompt = PromptTemplate(
    input_variables=["language", "code"],
    template=template
)

# 使用新的LCEL语法创建链（替代已弃用的LLMChain）
explain_chain = prompt | llm | StrOutputParser()

# 使用示例
code_example = """
def fibonacci(n):
    if n <= 1:
        return n
    return fibonacci(n-1) + fibonacci(n-2)
"""

result = explain_chain.invoke({
    "language": "Python",
    "code": code_example
})

print(result)

from langchain_community.llms import Tongyi
from langchain_core.prompts import PromptTemplate
from langchain_core.output_parsers import StrOutputParser
import os
from dotenv import load_dotenv

# 加载环境变量
load_dotenv()

# 检查API密钥
api_key = os.getenv("DASHSCOPE_API_KEY")
if not api_key or api_key == "your-actual-api-key":
    print("警告：请设置有效的DASHSCOPE_API_KEY环境变量")
    exit(1)

# 初始化模型
llm = Tongyi(
    model_name="qwen-turbo",
    temperature=0.5,  # 中等温度，平衡准确性和可读性
    max_tokens=500
)

# 创建提示模板
template = """请为以下文章生成一个简洁的摘要，要求：
1. 提取核心观点
2. 保留关键信息
3. 字数控制在100字以内

文章：
{article}

摘要："""

prompt = PromptTemplate(
    input_variables=["article"],
    template=template
)

# 创建链 - 使用LCEL语法替代LLMChain
summary_chain = prompt | llm | StrOutputParser()

# 使用示例
article = """
人工智能（AI）技术正在快速发展，深度学习、自然语言处理等技术已经
在各个领域得到广泛应用。大型语言模型如GPT、Claude等的出现，
使得AI能够理解和生成人类语言，极大提升了人机交互的体验。
然而，AI技术的发展也带来了一些挑战，包括数据隐私、算法偏见、
就业影响等问题。我们需要在推动创新的技术的同时，关注这些潜在风险，
确保AI技术的健康发展。
"""

result = summary_chain.invoke({"article": article})
print(result)