从调参到编排：LLM 应用优化的范式转移

# rag_context_engine.py — 最小可运行的 context engineering 示例
# 依赖: pip install langchain langchain-openai langchain-community chromadb

from langchain_openai import ChatOpenAI, OpenAIEmbeddings
from langchain_community.vectorstores import Chroma
from langchain_core.prompts import ChatPromptTemplate
from langchain_core.output_parsers import StrOutputParser
from langchain_core.runnables import RunnablePassthrough

# 1. 构建向量库（用几条模拟文档）
docs = [
    "Python 3.12 引入了更快的 CPython 解释器，f-string 支持嵌套表达式。",
    "JetBrains 的 DataSpell 专为数据科学家设计，支持 Jupyter 和 DataFrame 可视化。",
    "LangChain 0.2 重构了核心抽象，Runnable 接口统一了链式调用。",
    "RAG 系统的检索质量取决于分块策略和嵌入模型的选择，而非仅靠向量相似度。",
]

vectorstore = Chroma.from_texts(docs, embedding=OpenAIEmbeddings())
retriever = vectorstore.as_retriever(search_kwargs={"k": 2})

# 2. 结构化 prompt 模板——明确区分指令、上下文、问题
prompt = ChatPromptTemplate.from_messages([
    ("system", "你是一名技术顾问。只根据下方【参考资料】回答问题。"
               "如果资料中没有相关信息，直接说'我不确定'，不要编造。"),
    ("system", "【参考资料】\n{context}"),
    ("human", "{question}"),
])

# 3. 检索结果的格式化：编号 + 分隔，方便模型引用
def format_docs(docs):
    return "\n\n".join(f"[{i+1}] {d.page_content}" for i, d in enumerate(docs))

# 4. 组装链路
llm = ChatOpenAI(model="gpt-4o-mini", temperature=0)
chain = (
    {"context": retriever | format_docs, "question": RunnablePassthrough()}
    | prompt
    | llm
    | StrOutputParser()
)

# 5. 运行
if __name__ == "__main__":
    answer = chain.invoke("RAG 系统怎么提升检索质量？")
    print(answer)

# multi_agent_orchestrator.py — 三 agent 协作的最小实现
# 依赖: pip install openai

import os
from openai import OpenAI

client = OpenAI()  # 需要设置 OPENAI_API_KEY 环境变量

def call_agent(role: str, instruction: str, input_text: str) -> str:
    """通用 agent 调用：角色 + 指令 + 输入 → 输出"""
    response = client.chat.completions.create(
        model="gpt-4o-mini",
        temperature=0,
        messages=[
            {"role": "system", "content": f"你是{role}。{instruction}"},
            {"role": "user", "content": input_text},
        ],
    )
    return response.choices[0].message.content

# --- 三个 agent 的定义 ---
PLANNER_ROLE = "任务规划师"
PLANNER_INSTR = "将用户需求拆解为 2-4 个可执行的步骤，每步一句话描述。只输出步骤列表，不加解释。"

EXECUTOR_ROLE = "执行者"
EXECUTOR_INSTR = "根据给定的步骤和上下文，完成该步骤的任务。输出简洁的结果文本。"

REVIEWER_ROLE = "审查者"
REVIEWER_INSTR = ("对比最终结果与原始需求，判断是否满足。"
                   "如果满足，输出'PASS'；否则输出'FAIL: <具体原因>'。")

def orchestrate(user_request: str, max_retries: int = 2) -> str:
    """三 agent 编排主流程"""
    # Step 1: Planner 拆解任务
    steps_text = call_agent(PLANNER_ROLE, PLANNER_INSTR, user_request)
    print("=== 规划步骤 ===")
    print(steps_text)

    # Step 2: Executor 逐步执行
    context = f"原始需求: {user_request}\n步骤列表:\n{steps_text}"
    results = []
    for line in steps_text.strip().split("\n"):
        step = line.strip()
        if not step:
            continue
        result = call_agent(EXECUTOR_ROLE, EXECUTOR_INSTR,
                            f"当前步骤: {step}\n上下文: {context}")
        results.append(result)
        print(f"=== 执行 {step} ===\n{result}\n")

    # Step 3: Reviewer 审查
    final_output = "\n".join(results)
    review_input = f"原始需求: {user_request}\n最终结果:\n{final_output}"
    review = call_agent(REVIEWER_ROLE, REVIEWER_INSTR, review_input)
    print(f"=== 审查结果 ===\n{review}\n")

    # 如果审查失败且还有重试次数，重新规划
    if review.startswith("FAIL") and max_retries > 0:
        print("审查未通过，重新编排...")
        return orchestrate(
            f"{user_request}\n上次失败原因: {review}\n请调整步骤。",
            max_retries=max_retries - 1,
        )

    return final_output

if __name__ == "__main__":
    result = orchestrate("写一个 Python 函数，从 CSV 文件中筛选出金额大于 1000 的记录并按日期排序。")
    print("=== 最终输出 ===")
    print(result)