# 最简单的例子
from langchain_openai import ChatOpenAI

# 创建一个AI大脑
model = ChatOpenAI(model="gpt-3.5-turbo")

# 让AI回答问题
response = model.invoke("什么是Python？")
print(response.content)

from langchain.prompts import PromptTemplate

# 创建一个提示模板（就像填空题）
template = "你是一个{role}，请回答：{question}"

prompt = PromptTemplate(
    input_variables=["role", "question"],  # 需要填的空
    template=template
)

# 填空
formatted = prompt.format(
    role="Python老师",
    question="什么是变量？"
)

print(formatted)
# 输出：你是一个Python老师，请回答：什么是变量？

# ✅ 新版本（LangChain 1.2.0+）使用 LCEL
from langchain_core.output_parsers import StrOutputParser

# 创建一个链（使用 | 操作符连接）
chain = prompt | model | StrOutputParser()

# 一次性完成：填空 → 问AI → 得到答案
result = chain.invoke({
    "role": "数学老师",
    "question": "1+1等于几？"
})

print(result)  # 直接输出字符串

# ✅ 新版本（LangChain 1.2.0+）
from langchain_core.prompts import ChatPromptTemplate, MessagesPlaceholder
from langchain_core.runnables.history import RunnableWithMessageHistory
from langchain_community.chat_message_histories import ChatMessageHistory

# 创建记忆存储
store = {}

def get_session_history(session_id: str):
    if session_id not in store:
        store[session_id] = ChatMessageHistory()
    return store[session_id]

# 创建带记忆的对话链
prompt = ChatPromptTemplate.from_messages([
    ("system", "你是一个友好的助手"),
    MessagesPlaceholder(variable_name="history"),
    ("human", "{input}")
])
model = ChatOpenAI(model="gpt-4.1-mini")
chain = prompt | model

# 包装成带记忆的链
chain_with_history = RunnableWithMessageHistory(
    chain,
    get_session_history,
    input_messages_key="input",
    history_messages_key="history"
)

# 使用（会自动记住对话）
response1 = chain_with_history.invoke(
    {"input": "我叫张三"},
    config={"configurable": {"session_id": "user123"}}
)

response2 = chain_with_history.invoke(
    {"input": "我叫什么名字？"},
    config={"configurable": {"session_id": "user123"}}
)
# AI会回答：你叫张三

# ✅ 新版本（LangChain 1.2.0+）
from langchain.agents import create_tool_calling_agent, AgentExecutor
from langchain_core.tools import tool
from langchain import hub

# 定义工具（使用装饰器）
@tool
def calculator(expression: str) -> str:
    """计算数学表达式"""
    try:
        return str(eval(expression))
    except:
        return "计算错误"

tools = [calculator]

# 获取提示模板
prompt = hub.pull("hwchase17/openai-tools-agent")

# 创建 Agent
agent = create_tool_calling_agent(model, tools, prompt)
agent_executor = AgentExecutor(agent=agent, tools=tools, verbose=True)

# 使用
result = agent_executor.invoke({
    "input": "123 * 456 等于多少？"
})

print(result["output"])
# Agent会：
# 1. 思考：这是数学问题，需要用计算器
# 2. 使用计算器工具
# 3. 得到结果：56088
# 4. 回答用户

# ✅ 新版本（LangChain 1.2.0+）
from langchain_community.document_loaders import TextLoader
from langchain.text_splitter import RecursiveCharacterTextSplitter
from langchain_openai import OpenAIEmbeddings
from langchain_community.vectorstores import Chroma
from langchain_core.output_parsers import StrOutputParser
from langchain_core.prompts import ChatPromptTemplate
from langchain_core.runnables import RunnablePassthrough

# 步骤1：加载文档
loader = TextLoader("company_handbook.txt")
documents = loader.load()

# 步骤2：切分文档
splitter = RecursiveCharacterTextSplitter(chunk_size=1000, chunk_overlap=200)
chunks = splitter.split_documents(documents)

# 步骤3：向量化并存储
embeddings = OpenAIEmbeddings()
vectorstore = Chroma.from_documents(chunks, embeddings)
retriever = vectorstore.as_retriever()

# 步骤4：创建 RAG 链（使用 LCEL）
template = """基于以下上下文回答问题：

上下文：{context}

问题：{question}

答案："""

prompt = ChatPromptTemplate.from_template(template)

rag_chain = (
    {"context": retriever, "question": RunnablePassthrough()}
    | prompt
    | model
    | StrOutputParser()
)

# 步骤5：提问
answer = rag_chain.invoke("公司的年假政策是什么？")
print(answer)

# AI会：
# 1. 在文档中搜索相关内容
# 2. 基于找到的内容回答
# 3. 不会瞎编，因为有依据

# 1. 安装（在命令行运行）
# pip install langchain langchain-openai

# 2. 设置API密钥（在代码开头）
import os
os.environ["OPENAI_API_KEY"] = "你的API密钥"

# 3. 导入需要的类
from langchain_openai import ChatOpenAI
from langchain.prompts import ChatPromptTemplate
from langchain.schema.output_parser import StrOutputParser

# 4. 创建组件
model = ChatOpenAI(model="gpt-3.5-turbo")
prompt = ChatPromptTemplate.from_template("用简单的话解释：{topic}")
output_parser = StrOutputParser()

# 5. 组合成链（使用 | 操作符）
chain = prompt | model | output_parser

# 6. 使用
result = chain.invoke({"topic": "什么是机器学习"})
print(result)

# ❌ 错误
model = ChatOpenAI()  # 没有设置API密钥

# ✅ 正确
import os
os.environ["OPENAI_API_KEY"] = "sk-..."
model = ChatOpenAI()

# ❌ 错误
result = chain  # 忘记调用

# ✅ 正确
result = chain.invoke({"topic": "Python"})

# ❌ 错误
prompt = PromptTemplate(
    input_variables=["question"],  # 定义的是question
    template="解释{topic}"  # 但模板用的是topic
)

# ✅ 正确
prompt = PromptTemplate(
    input_variables=["topic"],  # 名字要一致
    template="解释{topic}"
)

chain = LLMChain(llm=model, prompt=prompt, verbose=True)
# verbose=True 会打印详细信息，方便调试

for chunk in model.stream("写一首诗"):
    print(chunk.content, end="", flush=True)
# 实时显示AI的输出，不用等全部生成完

from langchain.callbacks import get_openai_callback

with get_openai_callback() as cb:
    result = chain.invoke({"topic": "AI"})
    print(f"花费: ${cb.total_cost}")
    print(f"Token数: {cb.total_tokens}")

model = ChatOpenAI(
    max_retries=3,  # 失败自动重试3次
    timeout=30  # 30秒超时
)

# LLM - 文本补全
from langchain_openai import OpenAI
llm = OpenAI()
result = llm.invoke("从前有座山，")  # AI会接着写故事

# Chat Model - 对话
from langchain_openai import ChatOpenAI
chat = ChatOpenAI()
result = chat.invoke("你好")  # AI会回复你

# Embeddings - 向量化
from langchain_openai import OpenAIEmbeddings
embeddings = OpenAIEmbeddings()
vector = embeddings.embed_query("机器学习")  # 变成一串数字

from langchain.prompts import PromptTemplate

# 创建模板（就像填空题）
template = PromptTemplate(
    input_variables=["name", "job"],
    template="你好{name}，你是做{job}的吗？"
)

# 填空
question = template.format(name="张三", job="程序员")
# 结果：你好张三，你是做程序员的吗？

from langchain_community.chat_message_histories import ChatMessageHistory
from langchain_core.prompts import ChatPromptTemplate, MessagesPlaceholder
from langchain_core.runnables.history import RunnableWithMessageHistory

# 创建记忆存储
message_history = ChatMessageHistory()

# 保存对话
message_history.add_user_message("我叫张三")
message_history.add_ai_message("你好张三，很高兴认识你!")

# 查看记忆
print("对话历史:")
for msg in message_history.messages:
    print(f"{msg.type}: {msg.content}")

from langchain_core.output_parsers import StrOutputParser

# 1、创建大模型实例
chat_model = ChatOpenAI(model="gpt-4o-mini")
# 2、原始字符串模板
template = "桌上有{number}个苹果,四个桃子和 3 本书,一共有几个水果?"
prompt = PromptTemplate.from_template(template)
# 3、创建链式调用（使用LCEL语法,替代已弃用的LLMChain）
parser = StrOutputParser()
chain = prompt | chat_model | parser
# 4、调用chain,返回结果
result = chain.invoke({"number": 2})
print(result)

from langchain.agents import create_agent
from langchain_core.tools import tool

@tool
def get_weather(city: str) -> str:
    """获取指定城市的天气"""
    return f"{city}今天晴，气温 25 度"

agent = create_agent(
    "gpt-4o-mini",
    tools=[get_weather],
    system_prompt="你是一个有帮助的助手",
)

result = agent.invoke(
    {"messages": [{"role": "user", "content": "北京天气怎么样？"}]}
)
print(result["messages"][-1].content)

from langchain_community.document_loaders import TextLoader
from langchain.text_splitter import RecursiveCharacterTextSplitter

# 1. 加载文档
loader = TextLoader("document.txt")
docs = loader.load()

# 2. 切分文档
splitter = RecursiveCharacterTextSplitter(chunk_size=1000)
chunks = splitter.split_documents(docs)

# 3. 向量化存储
from langchain_community.vectorstores import Chroma
vectorstore = Chroma.from_documents(chunks, embeddings)

# 4. 搜索
results = vectorstore.similarity_search("机器学习")

# 伪代码示例
class BaseModel:
    def invoke(self, input): pass
    def batch(self, inputs): pass
    def stream(self, input): pass
    
class BaseRetriever:
    def get_relevant_documents(self, query): pass
    def aget_relevant_documents(self, query): pass  # 异步版本

# 组合示例（伪代码）
chain = (
    {"context": retriever, "question": RunnablePassthrough()}
    | prompt
    | model
    | output_parser
)

from langchain_openai import OpenAI

# 初始化模型
llm = OpenAI(
    model_name="gpt-3.5-turbo-instruct",
    temperature=0.7,  # 控制随机性
    max_tokens=256,   # 最大输出长度
)

# 单次调用
response = llm.invoke("解释什么是量子计算")

# 批量调用
responses = llm.batch([
    "什么是机器学习？",
    "什么是深度学习？"
])

# 流式输出
for chunk in llm.stream("写一首关于春天的诗"):
    print(chunk, end="", flush=True)

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

# 初始化聊天模型
chat = ChatOpenAI(
    model="gpt-4o-mini",  # 使用更经济的模型
    temperature=0.7
)

# 构建消息列表
messages = [
    SystemMessage(content="你是一个专业的Python编程助手"),
    HumanMessage(content="如何实现单例模式？"),
]

# 调用模型
response = chat.invoke(messages)
print(response.content)

# 多轮对话
messages.append(AIMessage(content=response.content))
messages.append(HumanMessage(content="能给个代码示例吗？"))
response = chat.invoke(messages)
print(response.content)

from langchain_openai import OpenAIEmbeddings

# 初始化
embeddings = OpenAIEmbeddings(
    model="text-embedding-3-small"
)

# 单个文本向量化
vector = embeddings.embed_query("机器学习是什么？")
print(f"向量维度: {len(vector)}")

# 批量向量化
texts = ["文本1", "文本2", "文本3"]
vectors = embeddings.embed_documents(texts)

# 计算相似度
from numpy import dot
from numpy.linalg import norm

def cosine_similarity(v1, v2):
    return dot(v1, v2) / (norm(v1) * norm(v2))

query_vec = embeddings.embed_query("深度学习")
doc_vec = embeddings.embed_query("神经网络")
similarity = cosine_similarity(query_vec, doc_vec)
print(f"相似度: {similarity}")

# 标准接口
class BaseLanguageModel:
    def invoke(self, input):
        """同步调用"""
        pass
    
    async def ainvoke(self, input):
        """异步调用"""
        pass
    
    def batch(self, inputs):
        """批量调用"""
        pass
    
    def stream(self, input):
        """流式输出"""
        pass

# 方式1：直接替换
from langchain_openai import ChatOpenAI
from langchain_anthropic import ChatAnthropic

# OpenAI模型
model = ChatOpenAI(model="gpt-4")

# 切换到Anthropic（接口完全相同）
model = ChatAnthropic(model="claude-3-opus")

# 方式2：配置化
def get_model(provider: str):
    if provider == "openai":
        return ChatOpenAI(model="gpt-4")
    elif provider == "anthropic":
        return ChatAnthropic(model="claude-3-opus")
    # ... 其他提供商

# OpenAI
export OPENAI_API_KEY="sk-..."
export OPENAI_API_BASE="https://api.openai.com/v1"

# Anthropic
export ANTHROPIC_API_KEY="sk-ant-..."

# HuggingFace
export HUGGINGFACEHUB_API_TOKEN="hf_..."

from langchain_openai import ChatOpenAI

model = ChatOpenAI(
    model="gpt-4",
    temperature=0.7,
    max_tokens=2000,
    timeout=60,  # 超时设置
    max_retries=3,  # 重试次数
    request_timeout=30,  # 请求超时
    streaming=True,  # 启用流式
    verbose=True,  # 调试模式
)

from langchain_community.callbacks import get_openai_callback
from tenacity import retry, stop_after_attempt, wait_exponential

# 成本追踪
with get_openai_callback() as cb:
    response = model.invoke("你好")
    print(f"Tokens: {cb.total_tokens}")
    print(f"Cost: ${cb.total_cost}")

# 自动重试
@retry(
    stop=stop_after_attempt(3),
    wait=wait_exponential(multiplier=1, min=4, max=10)
)
def call_model_with_retry(prompt):
    return model.invoke(prompt)

result = call_model_with_retry("你好")
print(result.content)

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

# 创建提示模板
prompt = ChatPromptTemplate.from_messages([
("system", "你是世界级的技术文档编写者。"),
("user", "{input}")
])
# 使用输出解析器
# output_parser = StrOutputParser()
output_parser = JsonOutputParser()
# 将其添加到上一个链中
# chain = prompt | llm
chain = prompt | llm | output_parser
# 调用它并提出同样的问题。答案是一个字符串，而不是ChatMessage
# chain.invoke({"input": "LangChain是什么?"})
chain.invoke({"input": "LangChain是什么? 用JSON格式回复，问题用question，回答用answer"})

from langchain_core.prompts import ChatPromptTemplate, SystemMessagePromptTemplate, HumanMessagePromptTemplate
from langchain_openai import ChatOpenAI
from langchain_core.output_parsers import StrOutputParser

# 初始化模型
llm = ChatOpenAI(model="gpt-4o-mini")

# 系统消息模板
system_template = "你是一个{expertise}专家，擅长{skill}。"
system_prompt = SystemMessagePromptTemplate.from_template(system_template)

# 用户消息模板
human_template = "{user_input}"
human_prompt = HumanMessagePromptTemplate.from_template(human_template)

# 组合
chat_prompt = ChatPromptTemplate.from_messages([
    system_prompt,
    human_prompt
])

# 创建输出解析器
output_parser = StrOutputParser()

# 创建链
chain = chat_prompt | llm | output_parser

# 使用
result = chain.invoke({
    "expertise": "数据分析",
    "skill": "Python和SQL",
    "user_input": "如何分析用户行为数据？"
})

print(result)

from langchain_core.prompts import FewShotPromptTemplate, PromptTemplate

# 定义示例
examples = [
    {
        "input": "happy",
        "output": "sad"
    },
    {
        "input": "tall",
        "output": "short"
    },
    {
        "input": "hot",
        "output": "cold"
    }
]

# 示例格式化模板
example_template = """
Input: {input}
Output: {output}
"""

example_prompt = PromptTemplate(
    input_variables=["input", "output"],
    template=example_template
)

# Few-shot模板
few_shot_prompt = FewShotPromptTemplate(
    examples=examples,
    example_prompt=example_prompt,
    prefix="给出以下词的反义词：",
    suffix="Input: {word}\nOutput:",
    input_variables=["word"]
)

# 使用
print(few_shot_prompt.format(word="big"))

from langchain.prompts.example_selector import SemanticSimilarityExampleSelector
from langchain_openai import OpenAIEmbeddings
from langchain_community.vectorstores import Chroma

# 创建示例选择器
example_selector = SemanticSimilarityExampleSelector.from_examples(
    examples,
    OpenAIEmbeddings(),
    Chroma,
    k=2  # 选择最相似的2个示例
)

# 使用选择器
few_shot_prompt = FewShotPromptTemplate(
    example_selector=example_selector,
    example_prompt=example_prompt,
    prefix="给出以下词的反义词：",
    suffix="Input: {word}\nOutput:",
    input_variables=["word"]
)

from langchain_core.output_parsers import PydanticOutputParser
from langchain_core.prompts import PromptTemplate
from pydantic import BaseModel, Field

# 定义数据模型
class Person(BaseModel):
    name: str = Field(description="人物姓名")
    age: int = Field(description="年龄")
    occupation: str = Field(description="职业")

# 创建解析器
parser = PydanticOutputParser(pydantic_object=Person)

# 获取格式说明
format_instructions = parser.get_format_instructions()

# 创建提示
prompt = PromptTemplate(
    template="提取以下文本中的人物信息：\n{text}\n\n{format_instructions}",
    input_variables=["text"],
    partial_variables={"format_instructions": format_instructions}
)

# 使用
text = "张三今年30岁，是一名软件工程师。"
formatted_prompt = prompt.format(text=text)
response = model.invoke(formatted_prompt)
person = parser.parse(response.content)

print(person.name)  # 张三
print(person.age)   # 30

from langchain_core.output_parsers import StructuredOutputParser, ResponseSchema

# 定义响应模式
response_schemas = [
    ResponseSchema(name="answer", description="问题的答案"),
    ResponseSchema(name="confidence", description="置信度（0-100）"),
    ResponseSchema(name="source", description="信息来源")
]

parser = StructuredOutputParser.from_response_schemas(response_schemas)

# 使用
format_instructions = parser.get_format_instructions()
# ... 构建提示并调用模型
result = parser.parse(response.content)
print(result["answer"])
print(result["confidence"])

from langchain.output_parsers import OutputFixingParser

# 原始解析器
base_parser = PydanticOutputParser(pydantic_object=Person)

# 包装修复解析器
fixing_parser = OutputFixingParser.from_llm(
    parser=base_parser,
    llm=model
)

# 即使输出格式有问题，也会尝试修复
try:
    result = fixing_parser.parse(malformed_output)
except Exception as e:
    print(f"解析失败: {e}")

# ❌ 不好的提示
"写点关于AI的东西"

# ✅ 好的提示
"""
请写一篇500字的文章，介绍人工智能在医疗领域的应用。
要求：
1. 包含至少3个具体应用场景
2. 使用通俗易懂的语言
3. 结构清晰，分段合理
"""

system_message = """
你是一个资深的Python开发工程师，拥有10年以上的开发经验。
你的特点：
- 代码风格遵循PEP 8规范
- 注重代码可读性和可维护性
- 善于使用设计模式解决问题
- 会考虑性能和安全性
"""

template = """
背景信息：
{context}

基于以上背景，请回答：
{question}

要求：
- 答案要基于提供的背景信息
- 如果背景信息不足，请明确指出
- 保持客观和准确
"""

template = """
请分析以下代码的时间复杂度：

{code}

请按以下格式输出：
1. 时间复杂度：O(?)
2. 空间复杂度：O(?)
3. 分析过程：[详细说明]
4. 优化建议：[如果有]
"""

template = """
问题：{question}

请按以下步骤思考：
1. 理解问题：明确问题要求
2. 分析思路：列出可能的解决方案
3. 选择方案：选择最优方案并说明理由
4. 详细步骤：给出具体实现步骤
5. 最终答案：给出最终结果

让我们一步步思考：
"""

from langchain.evaluation import load_evaluator

# 定义两个提示版本
prompt_a = PromptTemplate(template="简单提示：{question}")
prompt_b = PromptTemplate(template="详细提示：{question}\n请详细解释。")

# 评估器
evaluator = load_evaluator("criteria", criteria="helpfulness")

# 测试
test_questions = ["什么是机器学习？", "如何学习Python？"]
for question in test_questions:
    response_a = model.invoke(prompt_a.format(question=question))
    response_b = model.invoke(prompt_b.format(question=question))
    
    score_a = evaluator.evaluate_strings(
        prediction=response_a.content,
        input=question
    )
    score_b = evaluator.evaluate_strings(
        prediction=response_b.content,
        input=question
    )
    
    print(f"Prompt A: {score_a['score']}")
    print(f"Prompt B: {score_b['score']}")

from langchain.memory import ConversationBufferMemory
from langchain.chains import ConversationChain

# 创建记忆
memory = ConversationBufferMemory()

# 创建对话链
conversation = ConversationChain(
    llm=model,
    memory=memory,
    verbose=True
)

# 多轮对话
conversation.predict(input="你好，我叫张三")
conversation.predict(input="我的名字是什么？")  # 能记住之前的信息

# 查看历史
print(memory.load_memory_variables({}))

from langchain.memory import ConversationBufferWindowMemory

# 只保留最近3轮对话
memory = ConversationBufferWindowMemory(k=3)

conversation = ConversationChain(
    llm=model,
    memory=memory
)

# 第4轮对话时，第1轮会被丢弃

from langchain.memory import ConversationSummaryMemory

# 自动总结历史对话
memory = ConversationSummaryMemory(llm=model)

conversation = ConversationChain(
    llm=model,
    memory=memory
)

# 对话会被自动总结，节省Token
conversation.predict(input="介绍一下机器学习")
conversation.predict(input="深度学习呢？")

# 查看总结
print(memory.buffer)  # 显示总结后的内容

from langchain.memory import ConversationBufferMemory

memory = ConversationBufferMemory(
    memory_key="chat_history",  # 自定义键名
    return_messages=True,  # 返回消息对象而非字符串
    input_key="question",  # 输入键
    output_key="answer"  # 输出键
)

from langchain.memory import VectorStoreRetrieverMemory
from langchain_openai import OpenAIEmbeddings
from langchain_community.vectorstores import Chroma

# 创建向量存储
embeddings = OpenAIEmbeddings()
vectorstore = Chroma(embedding_function=embeddings)

# 创建检索器记忆
memory = VectorStoreRetrieverMemory(
    retriever=vectorstore.as_retriever(search_kwargs={"k": 3})
)

# 保存对话
memory.save_context(
    {"input": "我喜欢吃披萨"},
    {"output": "好的，我记住了你喜欢披萨"}
)

memory.save_context(
    {"input": "我的爱好是游泳"},
    {"output": "明白了，游泳是你的爱好"}
)

# 智能检索相关记忆
relevant_memories = memory.load_memory_variables(
    {"input": "我喜欢什么食物？"}
)
print(relevant_memories)  # 会检索到披萨相关的记忆

from langchain.memory import ConversationEntityMemory

# 自动提取和记忆实体
memory = ConversationEntityMemory(llm=model)

conversation = ConversationChain(
    llm=model,
    memory=memory
)

conversation.predict(input="张三在北京工作，是一名工程师")
conversation.predict(input="李四是张三的同事")

# 查看实体信息
print(memory.entity_store.store)
# 输出：{'张三': '在北京工作的工程师', '李四': '张三的同事'}

from langchain.memory import CombinedMemory, ConversationBufferMemory, ConversationSummaryMemory

# 组合多种记忆
short_term = ConversationBufferWindowMemory(k=5, memory_key="recent")
long_term = ConversationSummaryMemory(llm=model, memory_key="summary")

memory = CombinedMemory(memories=[short_term, long_term])

import json

# 保存记忆
memory_data = memory.load_memory_variables({})
with open("memory.json", "w") as f:
    json.dump(memory_data, f)

# 加载记忆
with open("memory.json", "r") as f:
    loaded_data = json.load(f)
    # 恢复记忆状态

# 清空记忆
memory.clear()

# 选择性删除
memory.chat_memory.messages = memory.chat_memory.messages[-10:]  # 只保留最近10条

from langchain_core.prompts import ChatPromptTemplate
from langchain_core.output_parsers import StrOutputParser
from langchain_openai import ChatOpenAI

# 创建组件
model = ChatOpenAI(model="gpt-3.5-turbo")
prompt = ChatPromptTemplate.from_template("为{product}写一句广告语")
output_parser = StrOutputParser()

# 使用 | 操作符组合（LCEL）
chain = prompt | model | output_parser

# 执行
result = chain.invoke({"product": "智能手表"})
print(result)

# 批量执行
results = chain.batch([
    {"product": "智能手表"},
    {"product": "无线耳机"}
])

# 流式输出
for chunk in chain.stream({"product": "智能音箱"}):
    print(chunk, end="", flush=True)

from langchain_core.runnables import RunnablePassthrough

# 步骤1：生成大纲
outline_prompt = ChatPromptTemplate.from_template("为{topic}生成文章大纲")
outline_chain = outline_prompt | model | StrOutputParser()

# 步骤2：根据大纲写文章
article_prompt = ChatPromptTemplate.from_template(
    "主题：{topic}\n大纲：{outline}\n\n请写一篇文章："
)

# 组合成顺序链
sequential_chain = (
    {"topic": RunnablePassthrough()}
    | RunnablePassthrough.assign(outline=outline_chain)
    | article_prompt
    | model
    | StrOutputParser()
)

# 执行
result = sequential_chain.invoke("人工智能的未来")
print(result)

from langchain_core.runnables import RunnableParallel

# 定义多个并行任务
summary_chain = (
    ChatPromptTemplate.from_template("总结：{text}")
    | model
    | StrOutputParser()
)

keywords_chain = (
    ChatPromptTemplate.from_template("提取关键词：{text}")
    | model
    | StrOutputParser()
)

sentiment_chain = (
    ChatPromptTemplate.from_template("分析情感：{text}")
    | model
    | StrOutputParser()
)

# 并行执行
parallel_chain = RunnableParallel(
    summary=summary_chain,
    keywords=keywords_chain,
    sentiment=sentiment_chain
)

# 执行
result = parallel_chain.invoke({"text": "今天天气真好，心情很愉快！"})
print(result)
# 输出：{'summary': '...', 'keywords': '...', 'sentiment': '...'}

from langchain_core.runnables import RunnableBranch

# 定义不同的处理链
physics_chain = (
    ChatPromptTemplate.from_template("作为物理学专家回答：{question}")
    | model
    | StrOutputParser()
)

math_chain = (
    ChatPromptTemplate.from_template("作为数学专家回答：{question}")
    | model
    | StrOutputParser()
)

general_chain = (
    ChatPromptTemplate.from_template("回答：{question}")
    | model
    | StrOutputParser()
)

# 创建分支
def route_question(input_dict):
    question = input_dict["question"].lower()
    if "物理" in question or "力" in question:
        return "physics"
    elif "数学" in question or "计算" in question:
        return "math"
    else:
        return "general"

branch_chain = RunnableBranch(
    (lambda x: route_question(x) == "physics", physics_chain),
    (lambda x: route_question(x) == "math", math_chain),
    general_chain  # 默认分支
)

# 执行
result = branch_chain.invoke({"question": "什么是牛顿第一定律？"})
print(result)

# 创建路由链
router_chain = MultiPromptChain.from_prompts(
    llm=model,
    prompt_infos=prompt_infos
)

# 自动路由到合适的专家
result = router_chain.invoke({"input": "什么是牛顿第一定律？"})  # 路由到physics

from langchain.chains import TransformChain

def transform_func(inputs: dict) -> dict:
    text = inputs["text"]
    # 自定义转换逻辑
    processed_text = text.upper()
    return {"processed_text": processed_text}

transform_chain = TransformChain(
    input_variables=["text"],
    output_variables=["processed_text"],
    transform=transform_func
)

# 与LLMChain组合
combined_chain = SequentialChain(
    chains=[transform_chain, llm_chain],
    input_variables=["text"],
    output_variables=["result"]
)

from langchain.chains.base import Chain
from typing import Dict, List

class CustomChain(Chain):
    """自定义Chain示例"""
    
    llm: Any
    prompt: PromptTemplate
    
    @property
    def input_keys(self) -> List[str]:
        """定义输入键"""
        return ["input_text"]
    
    @property
    def output_keys(self) -> List[str]:
        """定义输出键"""
        return ["output_text", "metadata"]
    
    def _call(self, inputs: Dict[str, str]) -> Dict[str, str]:
        """核心执行逻辑"""
        input_text = inputs["input_text"]
        
        # 自定义处理
        formatted_prompt = self.prompt.format(text=input_text)
        response = self.llm.invoke(formatted_prompt)
        
        # 返回结果
        return {
            "output_text": response.content,
            "metadata": {"length": len(response.content)}
        }

from langchain_core.runnables import RunnablePassthrough

# 方式1：管道操作符
chain = prompt | model | output_parser

# 方式2：并行执行
from langchain_core.runnables import RunnableParallel

chain = RunnableParallel(
    summary=summary_chain,
    keywords=keyword_chain
)

# 方式3：条件分支
from langchain_core.runnables import RunnableBranch

branch = RunnableBranch(
    (lambda x: len(x["text"]) < 100, short_chain),
    (lambda x: len(x["text"]) < 500, medium_chain),
    long_chain  # 默认
)

from langchain_core.runnables import RunnableWithFallbacks

# 主Chain失败时使用备用Chain
chain_with_fallback = primary_chain.with_fallbacks(
    [backup_chain_1, backup_chain_2]
)

# 自动重试
from tenacity import retry, stop_after_attempt

@retry(stop=stop_after_attempt(3))
def call_chain_with_retry(input_data):
    return chain.invoke(input_data)

from langchain.agents import create_openai_functions_agent, AgentExecutor
from langchain.tools import Tool
from langchain import hub

# 1. 定义工具
def search_tool(query: str) -> str:
    """搜索工具"""
    return f"搜索结果：{query}"

def calculator_tool(expression: str) -> str:
    """计算器工具"""
    return str(eval(expression))

tools = [
    Tool(
        name="Search",
        func=search_tool,
        description="用于搜索信息"
    ),
    Tool(
        name="Calculator",
        func=calculator_tool,
        description="用于数学计算"
    )
]

# 2. 获取提示模板
prompt = hub.pull("hwchase17/openai-functions-agent")

# 3. 创建Agent
agent = create_openai_functions_agent(
    llm=model,
    tools=tools,
    prompt=prompt
)

# 4. 创建执行器
agent_executor = AgentExecutor(
    agent=agent,
    tools=tools,
    verbose=True,
    max_iterations=5,  # 最大迭代次数
    max_execution_time=60,  # 最大执行时间
    handle_parsing_errors=True  # 处理解析错误
)

# 5. 执行
result = agent_executor.invoke({
    "input": "北京的人口是多少？然后计算人口的平方根"
})

from langchain.tools import tool

@tool
def get_word_length(word: str) -> int:
    """返回单词的长度"""
    return len(word)

@tool
def multiply(a: int, b: int) -> int:
    """计算两个数的乘积"""
    return a * b

from langchain.tools import Tool

def custom_function(input_str: str) -> str:
    # 自定义逻辑
    return f"处理结果：{input_str}"

tool = Tool(
    name="CustomTool",
    func=custom_function,
    description="这是一个自定义工具，用于..."
)

from langchain.tools import StructuredTool
from pydantic import BaseModel, Field

class SearchInput(BaseModel):
    query: str = Field(description="搜索查询")
    max_results: int = Field(default=5, description="最大结果数")

def search_function(query: str, max_results: int = 5) -> str:
    return f"搜索 '{query}'，返回 {max_results} 条结果"

search_tool = StructuredTool.from_function(
    func=search_function,
    name="Search",
    description="搜索工具",
    args_schema=SearchInput
)

from langchain.tools import WikipediaQueryRun, DuckDuckGoSearchRun
from langchain.utilities import WikipediaAPIWrapper

# Wikipedia工具
wikipedia = WikipediaQueryRun(api_wrapper=WikipediaAPIWrapper())

# DuckDuckGo搜索
search = DuckDuckGoSearchRun()

# Python REPL
from langchain.tools import PythonREPLTool
python_repl = PythonREPLTool()

tools = [wikipedia, search, python_repl]

agent_executor = AgentExecutor(
    agent=agent,
    tools=tools,
    verbose=True,
    
    # 迭代控制
    max_iterations=10,  # 最大迭代次数
    max_execution_time=120,  # 最大执行时间（秒）
    early_stopping_method="generate",  # 提前停止策略
    
    # 错误处理
    handle_parsing_errors=True,  # 自动处理解析错误
    return_intermediate_steps=True,  # 返回中间步骤
    
    # 记忆
    memory=memory,  # 添加记忆
    
    # 回调
    callbacks=[callback_handler]  # 添加回调
)

from langchain.agents import BaseSingleActionAgent
from typing import List, Tuple, Any

class CustomAgent(BaseSingleActionAgent):
    """自定义Agent"""
    
    tools: List[Tool]
    llm: Any
    
    @property
    def input_keys(self):
        return ["input"]
    
    def plan(
        self,
        intermediate_steps: List[Tuple[AgentAction, str]],
        **kwargs: Any
    ) -> Union[AgentAction, AgentFinish]:
        """决策逻辑"""
        # 自定义决策过程
        if len(intermediate_steps) > 5:
            return AgentFinish(
                return_values={"output": "达到最大步数"},
                log="停止执行"
            )
        
        # 选择工具
        tool_name = self._select_tool(kwargs["input"])
        return AgentAction(
            tool=tool_name,
            tool_input=kwargs["input"],
            log=f"使用工具: {tool_name}"
        )
    
    async def aplan(self, *args, **kwargs):
        """异步版本"""
        return self.plan(*args, **kwargs)

# 1. 启用详细输出
agent_executor = AgentExecutor(agent=agent, tools=tools, verbose=True)

# 2. 获取中间步骤
result = agent_executor.invoke(
    {"input": "问题"},
    return_intermediate_steps=True
)
for step in result["intermediate_steps"]:
    print(f"Action: {step[0]}")
    print(f"Observation: {step[1]}")

# 3. 使用回调追踪
from langchain.callbacks import StdOutCallbackHandler

agent_executor.invoke(
    {"input": "问题"},
    callbacks=[StdOutCallbackHandler()]
)

from langchain_community.document_loaders import TextLoader, PyPDFLoader, WebBaseLoader

# 1. 加载文本文件
loader = TextLoader("document.txt", encoding="utf-8")
documents = loader.load()

# 2. 加载PDF
pdf_loader = PyPDFLoader("document.pdf")
pages = pdf_loader.load_and_split()

# 3. 加载网页
web_loader = WebBaseLoader("https://example.com")
web_docs = web_loader.load()

# 4. 批量加载目录
from langchain_community.document_loaders import DirectoryLoader

loader = DirectoryLoader(
    "./docs",
    glob="**/*.md",  # 匹配所有markdown文件
    loader_cls=TextLoader
)
docs = loader.load()

from langchain.schema import Document

# Document对象包含两个主要属性
doc = Document(
    page_content="文档内容...",  # 文本内容
    metadata={  # 元数据
        "source": "document.pdf",
        "page": 1,
        "author": "张三"
    }
)

print(doc.page_content)
print(doc.metadata)

from langchain.text_splitter import RecursiveCharacterTextSplitter

# 创建分割器
splitter = RecursiveCharacterTextSplitter(
    chunk_size=1000,  # 每块大小
    chunk_overlap=200,  # 重叠部分
    length_function=len,  # 长度计算函数
    separators=["\n\n", "\n", " ", ""]  # 分隔符优先级
)

# 分割文档
chunks = splitter.split_documents(documents)

# 分割文本
texts = splitter.split_text("长文本...")

from langchain.text_splitter import TokenTextSplitter

# 按Token分割
token_splitter = TokenTextSplitter(
    chunk_size=512,  # Token数量
    chunk_overlap=50
)

chunks = token_splitter.split_documents(documents)

from langchain.text_splitter import Language, RecursiveCharacterTextSplitter

# Python代码分割
python_splitter = RecursiveCharacterTextSplitter.from_language(
    language=Language.PYTHON,
    chunk_size=500,
    chunk_overlap=50
)

# JavaScript代码分割
js_splitter = RecursiveCharacterTextSplitter.from_language(
    language=Language.JS,
    chunk_size=500
)

from langchain_community.vectorstores import Chroma
from langchain_openai import OpenAIEmbeddings

# 创建向量存储
embeddings = OpenAIEmbeddings()
vectorstore = Chroma.from_documents(
    documents=chunks,
    embedding=embeddings,
    persist_directory="./chroma_db"  # 持久化目录
)

# 相似度搜索
results = vectorstore.similarity_search(
    "机器学习是什么？",
    k=3  # 返回top 3
)

# 带分数的搜索
results_with_scores = vectorstore.similarity_search_with_score(
    "机器学习是什么？",
    k=3
)

for doc, score in results_with_scores:
    print(f"Score: {score}")
    print(f"Content: {doc.page_content}")

from langchain_community.vectorstores import Pinecone
import pinecone

# 初始化Pinecone
pinecone.init(
    api_key="your-api-key",
    environment="us-west1-gcp"
)

# 创建索引
index_name = "langchain-demo"
vectorstore = Pinecone.from_documents(
    documents=chunks,
    embedding=embeddings,
    index_name=index_name
)

# 搜索
results = vectorstore.similarity_search("查询文本", k=5)

# 1. 向量存储检索器
retriever = vectorstore.as_retriever(
    search_type="similarity",  # 相似度搜索
    search_kwargs={"k": 5}  # 返回5个结果
)

# 2. MMR检索（最大边际相关性）
retriever = vectorstore.as_retriever(
    search_type="mmr",  # 平衡相关性和多样性
    search_kwargs={"k": 5, "fetch_k": 20}
)

# 3. 相似度阈值检索
retriever = vectorstore.as_retriever(
    search_type="similarity_score_threshold",
    search_kwargs={"score_threshold": 0.8}  # 只返回分数>0.8的
)

from langchain.schema import BaseRetriever, Document

class CustomRetriever(BaseRetriever):
    """自定义检索器"""
    
    def _get_relevant_documents(self, query: str) -> List[Document]:
        """同步检索"""
        # 自定义检索逻辑
        return [Document(page_content="结果...")]
    
    async def _aget_relevant_documents(self, query: str) -> List[Document]:
        """异步检索"""
        return self._get_relevant_documents(query)

from langchain.retrievers import EnsembleRetriever

# 组合多个检索器
ensemble_retriever = EnsembleRetriever(
    retrievers=[vector_retriever, bm25_retriever],
    weights=[0.7, 0.3]  # 权重分配
)

results = ensemble_retriever.get_relevant_documents("查询")

from langchain_community.document_loaders import DirectoryLoader
from langchain.text_splitter import RecursiveCharacterTextSplitter
from langchain_openai import OpenAIEmbeddings
from langchain_community.vectorstores import Chroma

# 1. 加载文档
loader = DirectoryLoader(
    "./knowledge_base",
    glob="**/*.md",
    show_progress=True
)
documents = loader.load()

# 2. 文本分割
text_splitter = RecursiveCharacterTextSplitter(
    chunk_size=1000,
    chunk_overlap=200,
    separators=["\n\n", "\n", "。", "！", "？", "；", " ", ""]
)
chunks = text_splitter.split_documents(documents)

# 3. 添加元数据
for i, chunk in enumerate(chunks):
    chunk.metadata["chunk_id"] = i
    chunk.metadata["source_type"] = "markdown"

# 4. 向量化并存储
embeddings = OpenAIEmbeddings()
vectorstore = Chroma.from_documents(
    documents=chunks,
    embedding=embeddings,
    persist_directory="./vectorstore",
    collection_name="knowledge_base"
)

print(f"索引构建完成，共 {len(chunks)} 个文档块")

# 添加新文档
new_docs = loader.load_new_documents()
new_chunks = text_splitter.split_documents(new_docs)
vectorstore.add_documents(new_chunks)

# 删除文档
vectorstore.delete(ids=["doc_id_1", "doc_id_2"])

# 更新文档
vectorstore.update_document(doc_id="doc_id", document=new_doc)

from langchain.chains import RetrievalQA

# 创建检索器
retriever = vectorstore.as_retriever(
    search_type="similarity",
    search_kwargs={"k": 4}
)

# 创建RAG链
qa_chain = RetrievalQA.from_chain_type(
    llm=model,
    chain_type="stuff",  # 将所有文档塞入上下文
    retriever=retriever,
    return_source_documents=True  # 返回来源文档
)

# 查询
result = qa_chain.invoke({"query": "什么是机器学习？"})
print(result["result"])
print(result["source_documents"])

from langchain.retrievers.multi_query import MultiQueryRetriever

# 自动生成多个查询变体
multi_query_retriever = MultiQueryRetriever.from_llm(
    retriever=retriever,
    llm=model
)

# 单个问题生成多个查询，提高召回率
results = multi_query_retriever.get_relevant_documents(
    "机器学习的应用"
)

from langchain.retrievers import ContextualCompressionRetriever
from langchain.retrievers.document_compressors import LLMChainExtractor

# 使用LLM提取相关部分
compressor = LLMChainExtractor.from_llm(model)

compression_retriever = ContextualCompressionRetriever(
    base_compressor=compressor,
    base_retriever=retriever
)

# 只返回相关的文档片段
compressed_docs = compression_retriever.get_relevant_documents(
    "机器学习的定义"
)

from langchain.retrievers import ParentDocumentRetriever
from langchain.storage import InMemoryStore

# 小块检索，大块返回
parent_splitter = RecursiveCharacterTextSplitter(chunk_size=2000)
child_splitter = RecursiveCharacterTextSplitter(chunk_size=400)

store = InMemoryStore()

parent_retriever = ParentDocumentRetriever(
    vectorstore=vectorstore,
    docstore=store,
    child_splitter=child_splitter,
    parent_splitter=parent_splitter
)

from langchain.retrievers.document_compressors import CohereRerank

# 使用Cohere重排序
reranker = CohereRerank(model="rerank-english-v2.0", top_n=3)

rerank_retriever = ContextualCompressionRetriever(
    base_compressor=reranker,
    base_retriever=retriever
)

# 检索后重新排序，提高精确度
results = rerank_retriever.get_relevant_documents("查询")

from langchain.prompts import PromptTemplate

# 自定义RAG提示
template = """
你是一个专业的问答助手。请基于以下上下文回答问题。

上下文信息：
{context}

问题：{question}

回答要求：
1. 只基于上下文信息回答
2. 如果上下文中没有相关信息，明确说明
3. 引用具体的上下文片段
4. 保持客观和准确

回答：
"""

prompt = PromptTemplate(
    template=template,
    input_variables=["context", "question"]
)

qa_chain = RetrievalQA.from_chain_type(
    llm=model,
    retriever=retriever,
    chain_type_kwargs={"prompt": prompt}
)

# 策略1：固定大小 + 重叠
splitter1 = RecursiveCharacterTextSplitter(
    chunk_size=1000,
    chunk_overlap=200
)

# 策略2：语义分块
from langchain.text_splitter import SemanticChunker

splitter2 = SemanticChunker(
    embeddings=embeddings,
    breakpoint_threshold_type="percentile"  # 基于语义相似度分块
)

# 策略3：根据文档结构分块
from langchain.text_splitter import MarkdownHeaderTextSplitter

markdown_splitter = MarkdownHeaderTextSplitter(
    headers_to_split_on=[
        ("#", "Header 1"),
        ("##", "Header 2"),
        ("###", "Header 3"),
    ]
)