⚙️ LCEL 核心原理 ★

class Runnable(ABC, Generic[Input, Output]):
    """一个"工作单元"，可以被 invoke、batch、stream、transform 和组合。"""

    # 核心方法（所有 Runnable 都有）：
    # - invoke/ainvoke   : 把单个输入转成输出
    # - batch/abatch     : 高效地把多个输入转成输出（默认用线程池并行）
    # - stream/astream   : 流式输出单个输入的结果
    # - astream_log      : 流式输出 + 选定的中间结果

    # 内建优化：
    # - Batch: 默认用线程池并行跑多个 invoke()，子类可重写优化
    # - Async: 'a' 前缀的方法是异步版，默认降级用线程池跑同步版

    # 所有方法都接受可选的 config 参数（tags/metadata/callbacks/并发数）

    # 暴露 schema 信息：input_schema / output_schema / config_schema()

class Runnable:
    def pipe(self, *others, name=None):
        """把这个 Runnable 和 others 串联，返回 RunnableSequence。
        `a | b` 等价于 `a.pipe(b)`。"""
        return RunnableSequence(self, *others, name=name)

    def __or__(self, other):
        """重载 | 运算符，让 a | b 语法成立。"""
        return self.pipe(other)

# 所以这三行等价：
chain = prompt.pipe(model).pipe(parser)   # 显式
chain = prompt | model | parser           # 语法糖（最常用）
chain = RunnableSequence(prompt, model, parser)  # 最底层

chain = prompt | model | StrOutputParser()

# 1. 单次同步
chain.invoke({"question": "你好"})

# 2. 并发批量（自动用线程池并行！）
chain.batch([{"question": "你好"}, {"question": "再见"}, {"question": "谢谢"}])

# 3. 流式（逐字输出，自动透传 model 的 stream）
for chunk in chain.stream({"question": "讲个故事"}):
    print(chunk, end="", flush=True)

# 4. 异步（自动有 ainvoke/abatch/astream）
result = await chain.ainvoke({"question": "你好"})

class RunnableSequence(RunnableSerializable):
    """管道组合。把多个 Runnable 串联，前一个输出 = 后一个输入。

    自动实现：
    - invoke: 依次调用每一步
    - batch:  并发处理多个输入（每步内部并行）
    - stream: 逐级透传流式输出
    - ainvoke/abatch/astream: 异步版（自动降级到线程池）
    """
    first: Runnable      # 管道第一步
    middle: list[Runnable]  # 中间步骤
    last: Runnable       # 最后一步

# 字典字面量 = 并发执行
from langchain_core.runnables import RunnableLambda

# 三个函数并发跑，结果合并成 {"a": ..., "b": ..., "c": ...}
parallel = {
    "a": RunnableLambda(lambda x: x + 1),
    "b": RunnableLambda(lambda x: x * 2),
    "c": RunnableLambda(lambda x: x ** 2),
}
result = RunnableLambda(lambda x: x).pipe(parallel).invoke(5)
# 等价写法：
from langchain_core.runnables import RunnableParallel
result = RunnableParallel(a=..., b=..., c=...).invoke(5)
# → {"a": 6, "b": 10, "c": 25}

# retrieve 和 passthrough 并发，结果合并成 {"context": ..., "question": ...}
chain = (
    RunnableParallel({
        "context": retriever,           # 检索相关文档
        "question": RunnablePassthrough()  # 原样透传问题
    })
    | prompt
    | model
    | parser
)

from langchain_core.runnables import RunnablePassthrough

# RunnablePassthrough.assign 保留原 dict，并新增字段
chain = RunnablePassthrough.assign(
    summary=lambda d: f"问题长度={len(d['question'])}"
)
chain.invoke({"question": "你好吗"})  # 注意是 dict 输入
# → {"question": "你好吗", "summary": "问题长度=3"}  原字段还在！

# 对比 RunnableLambda：会替换整个输出
from langchain_core.runnables import RunnableLambda
RunnableLambda(lambda d: f"长度{len(d['question'])}").invoke({"question": "你好吗"})
# → "长度3"  （原 dict 没了）

# pip install langchain langchain-openai chromadb
# export OPENAI_API_KEY="sk-..."
from langchain_openai import ChatOpenAI
from langchain_core.prompts import ChatPromptTemplate
from langchain_core.output_parsers import StrOutputParser
from langchain_core.runnables import RunnablePassthrough, RunnableLambda

model = ChatOpenAI(model="gpt-4o-mini")

# ============ ① 最简单的 chain ============
prompt = ChatPromptTemplate.from_messages([
    ("system", "你是有帮助的助手。"),
    ("human", "{question}"),
])
chain = prompt | model | StrOutputParser()
print(chain.invoke({"question": "1+1=?"}))

# ============ ② 用 RunnableLambda 包普通函数 ============
# 任何函数都能变成 Runnable，从而加入管道
def word_count(text: str) -> str:
    return f"回答字数: {len(text)}"

chain2 = prompt | model | StrOutputParser() | RunnableLambda(word_count)
print(chain2.invoke({"question": "介绍Python"}))

# ============ ③ 经典 RAG 链 ============
# 假装有个 retriever（LC8 之后会讲真正的向量检索）
def fake_retrieve(question: str) -> str:
    return "（这是检索到的相关文档内容...）"

retriever = RunnableLambda(fake_retrieve)

rag_prompt = ChatPromptTemplate.from_messages([
    ("system", "根据以下上下文回答问题。\n上下文：{context}"),
    ("human", "{question}"),
])

# LCEL 的精髓在这一行！
rag_chain = (
    {                                 # ← 字典 = RunnableParallel（并发）
        "context": retriever,         #   检索
        "question": RunnablePassthrough(),  #   保留原问题
    }
    | rag_prompt                       # ← 拼 prompt
    | model                            # ← 调模型
    | StrOutputParser()                # ← 取文本
)

# 调用时只传问题字符串，retriever 和 passthrough 都接收它
print(rag_chain.invoke("Python是什么？"))
# 流式也自动可用：
# for chunk in rag_chain.stream("Python是什么？"): print(chunk, end="")

from langchain_core.runnables import chain  # 注意是小写的装饰器

@chain
def my_step(x: dict) -> str:
    """函数自动变成 Runnable，可放入管道。"""
    return f"处理了: {x['question']}"

# 等价于 RunnableLambda(my_step)，但写法更简洁
chain = prompt | my_step | model | StrOutputParser()

result = chain.invoke(
    {"question": "你好"},
    config={
        "tags": ["production", "v2"],         # 追踪标签
        "metadata": {"user_id": "u123"},       # 自定义元数据
        "max_concurrency": 5,                  # batch 并发上限
        # "callbacks": [...]                   # 回调（日志/metrics）
    }
)
# 这些信息会出现在 LangSmith 追踪里，便于调试和监控