结构化输出：Instructor + LiteLLM，以及 LangChain/LangGraph 方案

本文目标是说明如何让大模型强制输出指定 JSON 结构（Schema），并给出在不同框架下的具体示例。

---

一、Instructor + LiteLLM：让模型“强制输出指定 JSON 结构”

1) Instructor 在这里做了什么

Instructor 的核心能力是把一次普通 LLM 调用变成“带结构化约束的调用”，通常包含这几个环节：

• Schema 注入：把你定义的结构（常见为 Pydantic 模型）以合适形式注入提示词/工具定义，让模型明确必须产出哪些字段、字段类型、必填项、枚举范围等。
• 解析与校验：拿到模型原始输出后做 JSON 解析，并用 Pydantic 校验类型与约束。
• 失败自动重试：校验失败时，基于失败原因引导模型修正输出；最多重试 max_retries 次。
• 返回强类型对象：成功后直接返回 response_model 的实例，而不是纯文本或不受约束的 dict。

“强制”的关键不在于一句“请输出 JSON”，而在于输出 → 解析 → 校验 → 失败重试的闭环。

2) LiteLLM 在这里做了什么

LiteLLM 的价值是用统一接口调用多厂商模型。结合 from_litellm()（instructor 的适配器）后：

• 你依旧用 LiteLLM 的参数风格组织 kwargs（如 model、messages、temperature 等）。
• Instructor 接管 response_model 等结构化参数：负责约束输出、解析校验、失败重试，并把最终结果转换成 Schema 对象。

3) 典型调用方式（与你贴的片段对应）

你贴的逻辑本质是：

• kwargs["response_model"] = RouteSchema：指定返回必须符合 RouteSchema
• kwargs["max_retries"] = 3：最多自动修复 3 次
• result: RouteSchema = client.chat.completions.create(**kwargs)：返回值直接是 RouteSchema 实例

4) RouteSchema 示例（Pydantic）

from typing import Literal, Optional
from pydantic import BaseModel, Field

class RouteSchema(BaseModel):
    tool: Literal["search", "filesystem", "shell", "none"] = Field(..., description="要走的工具")
    reason: str = Field(..., description="为什么这么路由（给人看的简短理由）")
    query: Optional[str] = Field(None, description="如果 tool=search，需要搜索的关键词")

---

二、用 LangChain 或 LangGraph 能否实现同样效果？

可以。两者都能实现“强制输出指定 JSON 结构”，典型链路是：

定义结构（Pydantic/TypedDict） → 结构化输出/工具调用约束 → 解析校验失败自动修复或重试。

---

三、LangChain：结构化输出具体示例

示例 A：with_structured_output(Pydantic)（最接近 instructor 的体验）

from typing import Literal, Optional
from pydantic import BaseModel, Field
from langchain_openai import ChatOpenAI

class RouteSchema(BaseModel):
    tool: Literal["search", "filesystem", "shell", "none"] = Field(..., description="路由到哪个工具")
    reason: str = Field(..., description="简短理由")
    query: Optional[str] = Field(None, description="tool=search 时的关键词")

llm = ChatOpenAI(model="gpt-4o-mini", temperature=0)

# 关键：让 LLM 直接“返回 RouteSchema”
structured_llm = llm.with_structured_output(RouteSchema)

result: RouteSchema = structured_llm.invoke(
    "我需要查看当前目录有什么文件，应该调用哪个工具？"
)

print(result.tool, result.reason, result.query)

说明：result 是强类型对象；底层通常依赖工具调用/结构化输出协议并做校验。

示例 B：解析/校验失败自动修复（类似“失败就再试一次”）

from typing import Literal, Optional
from pydantic import BaseModel
from langchain_openai import ChatOpenAI
from langchain.output_parsers import PydanticOutputParser, OutputFixingParser
from langchain_core.prompts import ChatPromptTemplate

class RouteSchema(BaseModel):
    tool: Literal["search", "filesystem", "shell", "none"]
    reason: str
    query: Optional[str] = None

llm = ChatOpenAI(model="gpt-4o-mini", temperature=0)

parser = PydanticOutputParser(pydantic_object=RouteSchema)
fixing_parser = OutputFixingParser.from_llm(parser=parser, llm=llm)

prompt = ChatPromptTemplate.from_messages([
    ("system", "你是路由器，只能输出符合格式的 JSON。"),
    ("human", "{question}\n\n{format_instructions}"),
])

chain = prompt | llm | fixing_parser

result: RouteSchema = chain.invoke({
    "question": "我想读取 d:/github 下的文件列表，应该用哪个工具？",
    "format_instructions": parser.get_format_instructions(),
})

print(result)

---

四、LangGraph：在“图”里强制结构化输出并用于路由

LangGraph 的优势是：结构化输出不仅是“拿到 JSON”，还能直接用它来决定下一步节点（非常适合 router 场景）。

from typing import Literal, Optional, TypedDict
from pydantic import BaseModel, Field
from langchain_openai import ChatOpenAI
from langgraph.graph import StateGraph, END

class RouteSchema(BaseModel):
    tool: Literal["search", "filesystem", "shell", "none"] = Field(...)
    reason: str = Field(...)
    query: Optional[str] = None

class State(TypedDict, total=False):
    user_input: str
    route: RouteSchema

llm = ChatOpenAI(model="gpt-4o-mini", temperature=0)
router_llm = llm.with_structured_output(RouteSchema)

def router_node(state: State) -> State:
    route: RouteSchema = router_llm.invoke(
        f"根据用户需求选择 tool，并给出 reason。\n用户：{state['user_input']}"
    )
    return {"route": route}

def search_node(state: State) -> State:
    # 这里写你的 search 工具调用逻辑
    return {"user_input": state["user_input"]}

def shell_node(state: State) -> State:
    # 这里写你的 shell 工具调用逻辑
    return {"user_input": state["user_input"]}

def none_node(state: State) -> State:
    return {}

def route_selector(state: State) -> str:
    return state["route"].tool

g = StateGraph(State)
g.add_node("router", router_node)
g.add_node("search", search_node)
g.add_node("shell", shell_node)
g.add_node("none", none_node)

g.set_entry_point("router")
g.add_conditional_edges(
    "router",
    route_selector,
    {
        "search": "search",
        "shell": "shell",
        "none": "none",
        "filesystem": END,  # 示例：你也可以加 filesystem 节点
    },
)

g.add_edge("search", END)
g.add_edge("shell", END)
g.add_edge("none", END)

app = g.compile()

final_state = app.invoke({"user_input": "帮我列出当前目录文件"})
print(final_state["route"])

---

五、可靠性建议与常见坑（适用于三种方案）

• 用强约束类型提升一次通过率：Literal/Enum、范围约束、必填字段等。
• Schema 不要太深太复杂：嵌套太深会降低稳定性、提升重试概率。
• 保留 reason 字段：方便调试，也更利于失败重试时纠偏。
• 合理设置重试：一般 2～5 次足够，过大带来延迟与成本增加。
• 避免提示词冲突：例如同时要求“输出 Markdown 解释过程”与“只能输出 JSON”，容易导致解析失败。