# Java工程师写的Python（能跑，但不对）
class LLMService:
    def __init__(self):
        self._client = None

    def getClient(self):
        if self._client is None:
            self._client = OpenAI()
        return self._client

    def setClient(self, client):
        self._client = client

    def callLLM(self, prompt: str) -> str:
        response = self.getClient().chat.completions.create(
            model="gpt-4o",
            messages=[{"role": "user", "content": prompt}]
        )
        return response.choices[0].message.content

service = LLMService()
result = service.callLLM("你好")

# Python工程师写的同等代码
from openai import OpenAI

client = OpenAI()

def call_llm(prompt: str) -> str:
    response = client.chat.completions.create(
        model="gpt-4o",
        messages=[{"role": "user", "content": prompt}]
    )
    return response.choices[0].message.content

result = call_llm("你好")

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

# 三个组件用管道符连接，没有类，没有接口，没有实现
chain = ChatPromptTemplate.from_template("翻译成英文：{text}") | ChatOpenAI() | StrOutputParser()

result = chain.invoke({"text": "今天天气很好"})

# 加了Type Hints，mypy会在运行前报错
def process_documents(docs: list[str], max_tokens: int) -> list[dict]:
    ...

process_documents("单个字符串", "不是数字")  # mypy: Argument 1 has incompatible type

from pydantic import BaseModel, Field

class LLMRequest(BaseModel):
    prompt: str = Field(min_length=1, max_length=4000)
    temperature: float = Field(ge=0.0, le=2.0, default=0.7)
    model: str = "gpt-4o"

# 传错类型或超出范围，立刻抛出ValidationError，不会悄悄执行错误逻辑
request = LLMRequest(prompt="你好", temperature=5.0)  # ValidationError: temperature超出范围

def test_process_documents_empty_input():
    with pytest.raises(ValueError):
        process_documents([], max_tokens=100)

from pydantic import BaseModel, Field
from langchain_core.tools import tool

class SearchInput(BaseModel):
    query: str = Field(description="搜索关键词，尽量简洁明确")
    max_results: int = Field(default=5, ge=1, le=20, description="返回结果数量")
    date_filter: str | None = Field(default=None, description="日期过滤，格式 YYYY-MM-DD")

@tool(args_schema=SearchInput)
def search_web(query: str, max_results: int = 5, date_filter: str | None = None) -> list[dict]:
    """搜索互联网获取最新信息。当需要查询实时数据、新闻、当前价格时使用此工具。"""
    # 实现搜索逻辑
    ...

import asyncio
from openai import AsyncOpenAI, RateLimitError, APITimeoutError
import logging

logger = logging.getLogger(__name__)

async def call_llm_with_retry(
    prompt: str,
    max_retries: int = 3,
    base_delay: float = 1.0,
) -> str:
    client = AsyncOpenAI()

    for attempt in range(max_retries):
        try:
            response = await client.chat.completions.create(
                model="gpt-4o",
                messages=[{"role": "user", "content": prompt}],
                timeout=30.0,
            )
            return response.choices[0].message.content

        except openai.AuthenticationError:
            # ❌ 认证失败：不可重试！API Key无效或过期
            # 重试只会浪费时间，应该立刻抛出让调用方处理
            logger.error("API Key无效，请检查配置")
            raise  # 直接抛出，不重试

        except RateLimitError:
            # 429：触发限流，指数退避重试
            delay = base_delay * (2 ** attempt)
            logger.warning(f"Rate limit hit, retry {attempt+1}/{max_retries} after {delay}s")
            await asyncio.sleep(delay)

        except APITimeoutError:
            # 超时：快速重试一次
            logger.warning(f"Timeout on attempt {attempt+1}")
            if attempt == max_retries - 1:
                raise

        except openai.APIConnectionError as e:
            # 网络连接失败：可重试（比限流更常见）
            if attempt < max_retries - 1:
                wait = base_delay * (2 ** attempt)
                logger.warning(f"网络连接失败，{wait}s后重试: {e}")
                await asyncio.sleep(wait)
            else:
                raise

    raise RuntimeError(f"LLM call failed after {max_retries} retries")

from langchain_chroma import Chroma
from langchain_openai import OpenAIEmbeddings

def get_retriever(tenant_id: str, user_id: str):
    """
    多租户向量检索：通过 metadata 过滤隔离不同租户的数据
    和数据库的 WHERE tenant_id = ? 逻辑相同
    """
    vectorstore = Chroma(
        collection_name="documents",
        embedding_function=OpenAIEmbeddings(),
    )
    return vectorstore.as_retriever(
        search_kwargs={
            "filter": {"tenant_id": tenant_id},  # 等价于 SQL WHERE 子句
            "k": 5,
        }
    )

from langgraph.graph import StateGraph, END
from typing import TypedDict, Annotated
import operator

class AgentState(TypedDict):
    task: str
    subtasks: list[str]
    results: Annotated[list[str], operator.add]  # 多个worker写入，自动合并
    final_answer: str

def orchestrator(state: AgentState) -> AgentState:
    """主调度器：分析任务，拆分子任务"""
    # 调用LLM决定如何拆分任务
    subtasks = decompose_task(state["task"])
    return {"subtasks": subtasks}

def worker(state: AgentState) -> AgentState:
    """工作节点：执行单个子任务"""
    subtask = state["subtasks"][0]  # 取第一个未完成的子任务
    result = execute_subtask(subtask)
    return {"results": [result]}

def synthesizer(state: AgentState) -> AgentState:
    """汇总节点：合并所有结果生成最终答案"""
    final = synthesize_results(state["results"])
    return {"final_answer": final}

import os
from langsmith import traceable

os.environ["LANGCHAIN_TRACING_V2"] = "true"
os.environ["LANGCHAIN_API_KEY"] = "your-api-key"

@traceable(name="rag_pipeline")
async def rag_pipeline(question: str) -> str:
    docs = await retrieve_documents(question)       # 追踪：检索了哪些文档
    context = format_context(docs)
    answer = await generate_answer(question, context)  # 追踪：发了什么prompt，收到什么回复
    return answer

# 层次1：Type Hints（编写时检查，可选）
def call_llm(prompt: str, temperature: float = 0.7) -> str:
    ...

# 层次2：mypy 静态检查（pre-commit 或 CI 中运行）
# 运行：mypy src/
# 发现：call_llm(123)  → error: Argument 1 has incompatible type "int"; expected "str"

# 层次3：pydantic 运行时校验（数据进入系统时校验）
from pydantic import BaseModel, Field

class LLMRequest(BaseModel):
    prompt: str = Field(min_length=1, max_length=4000)
    temperature: float = Field(default=0.7, ge=0.0, le=2.0)

# 传入非法值，立刻在入口抛出 ValidationError，不会进入业务逻辑
request = LLMRequest(prompt="", temperature=5.0)
# ValidationError: prompt: min_length: 1, temperature: le: 2.0

# 阶段1：快速原型，无类型注解
def analyze(text, options):
    ...

# 阶段2：加基础类型注解
def analyze(text: str, options: dict) -> dict:
    ...

# 阶段3：精确类型（用 pydantic）
class AnalyzeOptions(BaseModel):
    language: str = "zh"
    max_length: int = 1000

def analyze(text: str, options: AnalyzeOptions) -> AnalyzeResult:
    ...

# 错误（所有调用共享同一个 tools_list）
def register_tool(tool, tools_list=[]):
    tools_list.append(tool)
    return tools_list

# 正确
def register_tool(tool, tools_list=None):
    if tools_list is None:
        tools_list = []
    tools_list.append(tool)
    return tools_list

# 错误（创建了协程对象但没有执行）
result = call_llm_async(prompt)  # 返回 <coroutine object>，什么都没发生
print(result)  # <coroutine object call_llm_async at 0x...>

# 正确
result = await call_llm_async(prompt)  # 真正执行，等待结果

# 错误（LLM 返回的 JSON 字段可能缺失）
content = response["choices"][0]["message"]["content"]  # KeyError 风险

# 正确（用 pydantic 在入口做防御）
from pydantic import BaseModel
class LLMResponse(BaseModel):
    content: str = ""
output = LLMResponse(**response["choices"][0]["message"])

# 错误（所有工具函数都引用同一个 i=2）
tools = [lambda: f"tool_{i}" for i in range(3)]
print(tools[0]())  # "tool_2"，不是 "tool_0"

# 正确
tools = [lambda i=i: f"tool_{i}" for i in range(3)]
print(tools[0]())  # "tool_0"

# 错误（你永远不知道 LLM 调用失败了）
try:
    result = call_llm(prompt)
except:
    pass  # 静默失败

# 正确
try:
    result = call_llm(prompt)
except Exception as e:
    logger.error(f"LLM call failed: {e}")
    raise  # 重新抛出，让调用方知道失败了