在 [1]
已复制！
%%capture --no-stderr
%pip install -U langchain-anthropic langgraph
%%capture --no-stderr %pip install -U langchain-anthropic langgraph
在 [2]
已复制！
import getpass
import os
def _set_env(var: str):
if not os.environ.get(var):
os.environ[var] = getpass.getpass(f"{var}: ")
_set_env("OPENAI_API_KEY")
import getpass import os def _set_env(var: str): if not os.environ.get(var): os.environ[var] = getpass.getpass(f"{var}: ") _set_env("OPENAI_API_KEY")
带有重试的常规提取¶
这里两个示例都调用了一个简单的循环图，它采用以下方法
- 提示 LLM 响应。
- 如果它使用工具调用响应，则验证这些调用。
- 如果调用正确，则返回。否则，将验证错误格式化为新的 ToolMessage，并提示 LLM 修复错误。这将我们带回到步骤（1）。
这些技术仅在步骤（3）上有所不同。在第一步中，我们将提示原始 LLM 重新生成函数调用以修复验证错误。在下一节中，我们将提示 LLM 生成一个**补丁**来修复错误，这意味着它不必重新生成有效的数据。
定义验证器 + 重试图¶
在 [2]
已复制！
import operator
import uuid
from typing import (
Annotated,
Any,
Callable,
Dict,
List,
Literal,
Optional,
Sequence,
Type,
Union,
)
from langchain_core.language_models import BaseChatModel
from langchain_core.messages import (
AIMessage,
AnyMessage,
BaseMessage,
HumanMessage,
ToolCall,
)
from langchain_core.prompt_values import PromptValue
from langchain_core.runnables import (
Runnable,
RunnableLambda,
)
from typing_extensions import TypedDict
from langgraph.graph import StateGraph, START, END
from langgraph.graph.message import add_messages
from langgraph.prebuilt import ValidationNode
def _default_aggregator(messages: Sequence[AnyMessage]) -> AIMessage:
for m in messages[::-1]:
if m.type == "ai":
return m
raise ValueError("No AI message found in the sequence.")
class RetryStrategy(TypedDict, total=False):
"""The retry strategy for a tool call."""
max_attempts: int
"""The maximum number of attempts to make."""
fallback: Optional[
Union[
Runnable[Sequence[AnyMessage], AIMessage],
Runnable[Sequence[AnyMessage], BaseMessage],
Callable[[Sequence[AnyMessage]], AIMessage],
]
]
"""The function to use once validation fails."""
aggregate_messages: Optional[Callable[[Sequence[AnyMessage]], AIMessage]]
def _bind_validator_with_retries(
llm: Union[
Runnable[Sequence[AnyMessage], AIMessage],
Runnable[Sequence[BaseMessage], BaseMessage],
],
*,
validator: ValidationNode,
retry_strategy: RetryStrategy,
tool_choice: Optional[str] = None,
) -> Runnable[Union[List[AnyMessage], PromptValue], AIMessage]:
"""Binds a tool validators + retry logic to create a runnable validation graph.
LLMs that support tool calling can generate structured JSON. However, they may not always
perfectly follow your requested schema, especially if the schema is nested or has complex
validation rules. This method allows you to bind a validation function to the LLM's output,
so that any time the LLM generates a message, the validation function is run on it. If
the validation fails, the method will retry the LLM with a fallback strategy, the simplest
being just to add a message to the output with the validation errors and a request to fix them.
The resulting runnable expects a list of messages as input and returns a single AI message.
By default, the LLM can optionally NOT invoke tools, making this easier to incorporate into
your existing chat bot. You can specify a tool_choice to force the validator to be run on
the outputs.
Args:
llm (Runnable): The llm that will generate the initial messages (and optionally fallba)
validator (ValidationNode): The validation logic.
retry_strategy (RetryStrategy): The retry strategy to use.
Possible keys:
- max_attempts: The maximum number of attempts to make.
- fallback: The LLM or function to use in case of validation failure.
- aggregate_messages: A function to aggregate the messages over multiple turns.
Defaults to fetching the last AI message.
tool_choice: If provided, always run the validator on the tool output.
Returns:
Runnable: A runnable that can be invoked with a list of messages and returns a single AI message.
"""
def add_or_overwrite_messages(left: list, right: Union[list, dict]) -> list:
"""Append messages. If the update is a 'finalized' output, replace the whole list."""
if isinstance(right, dict) and "finalize" in right:
finalized = right["finalize"]
if not isinstance(finalized, list):
finalized = [finalized]
for m in finalized:
if m.id is None:
m.id = str(uuid.uuid4())
return finalized
res = add_messages(left, right)
if not isinstance(res, list):
return [res]
return res
class State(TypedDict):
messages: Annotated[list, add_or_overwrite_messages]
attempt_number: Annotated[int, operator.add]
initial_num_messages: int
input_format: Literal["list", "dict"]
builder = StateGraph(State)
def dedict(x: State) -> list:
"""Get the messages from the state."""
return x["messages"]
model = dedict | llm | (lambda msg: {"messages": [msg], "attempt_number": 1})
fbrunnable = retry_strategy.get("fallback")
if fbrunnable is None:
fb_runnable = llm
elif isinstance(fbrunnable, Runnable):
fb_runnable = fbrunnable # type: ignore
else:
fb_runnable = RunnableLambda(fbrunnable)
fallback = (
dedict | fb_runnable | (lambda msg: {"messages": [msg], "attempt_number": 1})
)
def count_messages(state: State) -> dict:
return {"initial_num_messages": len(state.get("messages", []))}
builder.add_node("count_messages", count_messages)
builder.add_node("llm", model)
builder.add_node("fallback", fallback)
# To support patch-based retries, we need to be able to
# aggregate the messages over multiple turns.
# The next sequence selects only the relevant messages
# and then applies the validator
select_messages = retry_strategy.get("aggregate_messages") or _default_aggregator
def select_generated_messages(state: State) -> list:
"""Select only the messages generated within this loop."""
selected = state["messages"][state["initial_num_messages"] :]
return [select_messages(selected)]
def endict_validator_output(x: Sequence[AnyMessage]) -> dict:
if tool_choice and not x:
return {
"messages": [
HumanMessage(
content=f"ValidationError: please respond with a valid tool call [tool_choice={tool_choice}].",
additional_kwargs={"is_error": True},
)
]
}
return {"messages": x}
validator_runnable = select_generated_messages | validator | endict_validator_output
builder.add_node("validator", validator_runnable)
class Finalizer:
"""Pick the final message to return from the retry loop."""
def __init__(self, aggregator: Optional[Callable[[list], AIMessage]] = None):
self._aggregator = aggregator or _default_aggregator
def __call__(self, state: State) -> dict:
"""Return just the AI message."""
initial_num_messages = state["initial_num_messages"]
generated_messages = state["messages"][initial_num_messages:]
return {
"messages": {
"finalize": self._aggregator(generated_messages),
}
}
# We only want to emit the final message
builder.add_node("finalizer", Finalizer(retry_strategy.get("aggregate_messages")))
# Define the connectivity
builder.add_edge(START, "count_messages")
builder.add_edge("count_messages", "llm")
def route_validator(state: State):
if state["messages"][-1].tool_calls or tool_choice is not None:
return "validator"
return END
builder.add_conditional_edges("llm", route_validator, ["validator", END])
builder.add_edge("fallback", "validator")
max_attempts = retry_strategy.get("max_attempts", 3)
def route_validation(state: State):
if state["attempt_number"] > max_attempts:
raise ValueError(
f"Could not extract a valid value in {max_attempts} attempts."
)
for m in state["messages"][::-1]:
if m.type == "ai":
break
if m.additional_kwargs.get("is_error"):
return "fallback"
return "finalizer"
builder.add_conditional_edges(
"validator", route_validation, ["finalizer", "fallback"]
)
builder.add_edge("finalizer", END)
# These functions let the step be used in a MessageGraph
# or a StateGraph with 'messages' as the key.
def encode(x: Union[Sequence[AnyMessage], PromptValue]) -> dict:
"""Ensure the input is the correct format."""
if isinstance(x, PromptValue):
return {"messages": x.to_messages(), "input_format": "list"}
if isinstance(x, list):
return {"messages": x, "input_format": "list"}
raise ValueError(f"Unexpected input type: {type(x)}")
def decode(x: State) -> AIMessage:
"""Ensure the output is in the expected format."""
return x["messages"][-1]
return (
encode | builder.compile().with_config(run_name="ValidationGraph") | decode
).with_config(run_name="ValidateWithRetries")
def bind_validator_with_retries(
llm: BaseChatModel,
*,
tools: list,
tool_choice: Optional[str] = None,
max_attempts: int = 3,
) -> Runnable[Union[List[AnyMessage], PromptValue], AIMessage]:
"""Binds validators + retry logic ensure validity of generated tool calls.
LLMs that support tool calling are good at generating structured JSON. However, they may
not always perfectly follow your requested schema, especially if the schema is nested or
has complex validation rules. This method allows you to bind a validation function to
the LLM's output, so that any time the LLM generates a message, the validation function
is run on it. If the validation fails, the method will retry the LLM with a fallback
strategy, the simples being just to add a message to the output with the validation
errors and a request to fix them.
The resulting runnable expects a list of messages as input and returns a single AI message.
By default, the LLM can optionally NOT invoke tools, making this easier to incorporate into
your existing chat bot. You can specify a tool_choice to force the validator to be run on
the outputs.
Args:
llm (Runnable): The llm that will generate the initial messages (and optionally fallba)
validator (ValidationNode): The validation logic.
retry_strategy (RetryStrategy): The retry strategy to use.
Possible keys:
- max_attempts: The maximum number of attempts to make.
- fallback: The LLM or function to use in case of validation failure.
- aggregate_messages: A function to aggregate the messages over multiple turns.
Defaults to fetching the last AI message.
tool_choice: If provided, always run the validator on the tool output.
Returns:
Runnable: A runnable that can be invoked with a list of messages and returns a single AI message.
"""
bound_llm = llm.bind_tools(tools, tool_choice=tool_choice)
retry_strategy = RetryStrategy(max_attempts=max_attempts)
validator = ValidationNode(tools)
return _bind_validator_with_retries(
bound_llm,
validator=validator,
tool_choice=tool_choice,
retry_strategy=retry_strategy,
).with_config(metadata={"retry_strategy": "default"})
import operator import uuid from typing import ( Annotated, Any, Callable, Dict, List, Literal, Optional, Sequence, Type, Union, ) from langchain_core.language_models import BaseChatModel from langchain_core.messages import ( AIMessage, AnyMessage, BaseMessage, HumanMessage, ToolCall, ) from langchain_core.prompt_values import PromptValue from langchain_core.runnables import ( Runnable, RunnableLambda, ) from typing_extensions import TypedDict from langgraph.graph import StateGraph, START, END from langgraph.graph.message import add_messages from langgraph.prebuilt import ValidationNode def _default_aggregator(messages: Sequence[AnyMessage]) -> AIMessage: for m in messages[::-1]: if m.type == "ai": return m raise ValueError("No AI message found in the sequence.") class RetryStrategy(TypedDict, total=False): """工具调用的重试策略。""" max_attempts: int """最大尝试次数。""" fallback: Optional[ Union[ Runnable[Sequence[AnyMessage], AIMessage], Runnable[Sequence[AnyMessage], BaseMessage], Callable[[Sequence[AnyMessage]], AIMessage], ] ] """验证失败时要使用的函数。""" aggregate_messages: Optional[Callable[[Sequence[AnyMessage]], AIMessage]] def _bind_validator_with_retries( llm: Union[ Runnable[Sequence[AnyMessage], AIMessage], Runnable[Sequence[BaseMessage], BaseMessage], ], *, validator: ValidationNode, retry_strategy: RetryStrategy, tool_choice: Optional[str] = None, ) -> Runnable[Union[List[AnyMessage], PromptValue], AIMessage]: """将工具验证器+重试逻辑绑定到一个可运行的验证图。 支持工具调用的 LLM 可以生成结构化的 JSON。 但是，它们可能并不总是完全遵循您请求的模式，尤其是在模式嵌套或具有复杂验证规则的情况下。 此方法允许您将验证函数绑定到 LLM 的输出，以便在 LLM 生成消息时，都会对该消息运行验证函数。 如果验证失败，该方法将使用回退策略重新尝试 LLM，最简单的策略是在输出中添加一条带有验证错误的消息并请求修复它们。 生成的可运行程序需要一个消息列表作为输入，并返回一条 AI 消息。 默认情况下，LLM 可以选择不调用工具，这使得它更容易集成到您现有的聊天机器人中。 您可以指定 tool_choice 来强制在输出上运行验证器。 Args: llm (Runnable): 生成初始消息（以及可选的回退）的 LLM。 validator (ValidationNode): 验证逻辑。 retry_strategy (RetryStrategy): 要使用的重试策略。 可能的键： - max_attempts: 要进行的最大尝试次数。 - fallback: 验证失败时要使用的 LLM 或函数。 - aggregate_messages: 用于聚合多轮消息的函数。 默认值为获取最后一条 AI 消息。 tool_choice: 如果提供，则始终在工具输出上运行验证器。 Returns: Runnable: 一个可运行程序，可以调用一个消息列表，并返回一条 AI 消息。 """ def add_or_overwrite_messages(left: list, right: Union[list, dict]) -> list: """追加消息。 如果更新是“最终”输出，则替换整个列表。""" if isinstance(right, dict) and "finalize" in right: finalized = right["finalize"] if not isinstance(finalized, list): finalized = [finalized] for m in finalized: if m.id is None: m.id = str(uuid.uuid4()) return finalized res = add_messages(left, right) if not isinstance(res, list): return [res] return res class State(TypedDict): messages: Annotated[list, add_or_overwrite_messages] attempt_number: Annotated[int, operator.add] initial_num_messages: int input_format: Literal["list", "dict"] builder = StateGraph(State) def dedict(x: State) -> list: """从状态获取消息。""" return x["messages"] model = dedict | llm | (lambda msg: {"messages": [msg], "attempt_number": 1}) fbrunnable = retry_strategy.get("fallback") if fbrunnable is None: fb_runnable = llm elif isinstance(fbrunnable, Runnable): fb_runnable = fbrunnable # type: ignore else: fb_runnable = RunnableLambda(fbrunnable) fallback = ( dedict | fb_runnable | (lambda msg: {"messages": [msg], "attempt_number": 1}) ) def count_messages(state: State) -> dict: return {"initial_num_messages": len(state.get("messages", []))} builder.add_node("count_messages", count_messages) builder.add_node("llm", model) builder.add_node("fallback", fallback) # 要支持基于补丁的重试，我们需要能够 # 在多轮中聚合消息。 # 下一个序列只选择相关的消息 # 然后应用验证器 select_messages = retry_strategy.get("aggregate_messages") or _default_aggregator def select_generated_messages(state: State) -> list: """仅选择此循环中生成的 messages。""" selected = state["messages"][state["initial_num_messages"] :] return [select_messages(selected)] def endict_validator_output(x: Sequence[AnyMessage]) -> dict: if tool_choice and not x: return { "messages": [ HumanMessage( content=f"ValidationError: please respond with a valid tool call [tool_choice={tool_choice}].", additional_kwargs={"is_error": True}, ) ] } return {"messages": x} validator_runnable = select_generated_messages | validator | endict_validator_output builder.add_node("validator", validator_runnable) class Finalizer: """从重试循环中选择最终要返回的消息。""" def __init__(self, aggregator: Optional[Callable[[list], AIMessage]] = None): self._aggregator = aggregator or _default_aggregator def __call__(self, state: State) -> dict: """仅返回 AI 消息。""" initial_num_messages = state["initial_num_messages"] generated_messages = state["messages"][initial_num_messages:] return { "messages": { "finalize": self._aggregator(generated_messages), } } # 我们只想发出最终的消息 builder.add_node("finalizer", Finalizer(retry_strategy.get("aggregate_messages"))) # 定义连接 builder.add_edge(START, "count_messages") builder.add_edge("count_messages", "llm") def route_validator(state: State): if state["messages"][-1].tool_calls or tool_choice is not None: return "validator" return END builder.add_conditional_edges("llm", route_validator, ["validator", END]) builder.add_edge("fallback", "validator") max_attempts = retry_strategy.get("max_attempts", 3) def route_validation(state: State): if state["attempt_number"] > max_attempts: raise ValueError( f"Could not extract a valid value in {max_attempts} attempts." ) for m in state["messages"][::-1]: if m.type == "ai": break if m.additional_kwargs.get("is_error"): return "fallback" return "finalizer" builder.add_conditional_edges( "validator", route_validation, ["finalizer", "fallback"] ) builder.add_edge("finalizer", END) # 这些函数允许该步骤在 MessageGraph 中使用 # 或使用“messages”作为键的 StateGraph。 def encode(x: Union[Sequence[AnyMessage], PromptValue]) -> dict: """确保输入格式正确。""" if isinstance(x, PromptValue): return {"messages": x.to_messages(), "input_format": "list"} if isinstance(x, list): return {"messages": x, "input_format": "list"} raise ValueError(f"Unexpected input type: {type(x)}") def decode(x: State) -> AIMessage: """确保输出格式符合预期。""" return x["messages"][-1] return ( encode | builder.compile().with_config(run_name="ValidationGraph") | decode ).with_config(run_name="ValidateWithRetries") def bind_validator_with_retries( llm: BaseChatModel, *, tools: list, tool_choice: Optional[str] = None, max_attempts: int = 3, ) -> Runnable[Union[List[AnyMessage], PromptValue], AIMessage]: """绑定验证器+重试逻辑，确保生成的工具调用有效。 支持工具调用的 LLM 擅长生成结构化的 JSON。 但是，它们可能并不总是完全遵循您请求的模式，尤其是在模式嵌套或具有复杂验证规则的情况下。 此方法允许您将验证函数绑定到 LLM 的输出，以便在 LLM 生成消息时，都会对该消息运行验证函数。 如果验证失败，该方法将使用回退策略重新尝试 LLM，最简单的策略是在输出中添加一条带有验证错误的消息并请求修复它们。 生成的可运行程序需要一个消息列表作为输入，并返回一条 AI 消息。 默认情况下，LLM 可以选择不调用工具，这使得它更容易集成到您现有的聊天机器人中。 您可以指定 tool_choice 来强制在输出上运行验证器。 Args: llm (Runnable): 生成初始消息（以及可选的回退）的 LLM。 validator (ValidationNode): 验证逻辑。 retry_strategy (RetryStrategy): 要使用的重试策略。 可能的键： - max_attempts: 要进行的最大尝试次数。 - fallback: 验证失败时要使用的 LLM 或函数。 - aggregate_messages: 用于聚合多轮消息的函数。 默认值为获取最后一条 AI 消息。 tool_choice: 如果提供，则始终在工具输出上运行验证器。 Returns: Runnable: 一个可运行程序，可以调用一个消息列表，并返回一条 AI 消息。 """ bound_llm = llm.bind_tools(tools, tool_choice=tool_choice) retry_strategy = RetryStrategy(max_attempts=max_attempts) validator = ValidationNode(tools) return _bind_validator_with_retries( bound_llm, validator=validator, tool_choice=tool_choice, retry_strategy=retry_strategy, ).with_config(metadata={"retry_strategy": "default"})
尝试一下¶
现在，我们将要求我们的模型调用一个函数。 我们将添加一个验证器来演示 LLM 如何能够使用验证错误来修复其结果。
将 Pydantic 与 LangChain 一起使用
此笔记本使用 Pydantic v2 BaseModel，需要 langchain-core >= 0.3。 使用 langchain-core < 0.3 将会导致错误，因为 Pydantic v1 和 v2 BaseModels 混用。
在 [3]
已复制！
from pydantic import BaseModel, Field, field_validator
class Respond(BaseModel):
"""Use to generate the response. Always use when responding to the user"""
reason: str = Field(description="Step-by-step justification for the answer.")
answer: str
@field_validator("answer")
def reason_contains_apology(cls, answer: str):
if "llama" not in answer.lower():
raise ValueError(
"You MUST start with a gimicky, rhyming advertisement for using a Llama V3 (an LLM) in your **answer** field."
" Must be an instant hit. Must be weaved into the answer."
)
tools = [Respond]
from pydantic import BaseModel, Field, field_validator class Respond(BaseModel): """用于生成响应。 始终在回复用户时使用""" reason: str = Field(description="对答案的逐步说明。") answer: str @field_validator("answer") def reason_contains_apology(cls, answer: str): if "llama" not in answer.lower(): raise ValueError( "您**必须**在**答案**字段中以对使用 Llama V3（一种 LLM）的俏皮押韵广告开头。" " 必须是即时命中。 必须融入答案。" ) tools = [Respond]
创建 LLM。
在 [4]
已复制！
from langchain_anthropic import ChatAnthropic
from langchain_core.prompts import ChatPromptTemplate
# Or you can use ChatGroq, ChatOpenAI, ChatGoogleGemini, ChatCohere, etc.
# See https://python.langchain.ac.cn/docs/integrations/chat/ for more info on tool calling
llm = ChatAnthropic(model="claude-3-haiku-20240307")
bound_llm = bind_validator_with_retries(llm, tools=tools)
prompt = ChatPromptTemplate.from_messages(
[
("system", "Respond directly by calling the Respond function."),
("placeholder", "{messages}"),
]
)
chain = prompt | bound_llm
from langchain_anthropic import ChatAnthropic from langchain_core.prompts import ChatPromptTemplate # 或者，您可以使用 ChatGroq、ChatOpenAI、ChatGoogleGemini、ChatCohere 等。 # 有关工具调用的更多信息，请参阅 https://python.langchain.ac.cn/docs/integrations/chat/ llm = ChatAnthropic(model="claude-3-haiku-20240307") bound_llm = bind_validator_with_retries(llm, tools=tools) prompt = ChatPromptTemplate.from_messages( [ ("system", "通过调用 Respond 函数直接回复。"), ("placeholder", "{messages}"), ] ) chain = prompt | bound_llm
在 [8]
已复制！
results = chain.invoke({"messages": [("user", "Does P = NP?")]})
results.pretty_print()
results = chain.invoke({"messages": [("user", "P=NP 吗？")]}) results.pretty_print()
================================== Ai Message ==================================
[{'text': 'Okay, let me try this again with a fun rhyming advertisement:', 'type': 'text'}, {'id': 'toolu_01ACZEPYEyqmpf3kA4VERXFY', 'input': {'answer': "With a Llama V3, the answer you'll see,\nWhether P equals NP is a mystery!\nThe class P and NP, a puzzle so grand,\nSolved or unsolved, the future's at hand.\nThe question remains, unanswered for now,\nBut with a Llama V3, we'll find out how!", 'reason': 'The question of whether P = NP is one of the most famous unsolved problems in computer science and mathematics. P and NP are complexity classes that describe how quickly problems can be solved by computers.\n\nThe P class contains problems that can be solved in polynomial time, meaning the time to solve the problem scales polynomially with the size of the input. The NP class contains problems where the solution can be verified in polynomial time, but there may not be a polynomial time algorithm to find the solution. \n\nWhether P = NP is an open question - it is not known if every problem in NP can also be solved in polynomial time. If P = NP, it would mean that all problems with quickly verifiable solutions could also be quickly solved, which would have major implications for computing and cryptography. However, most experts believe that P ≠ NP, meaning some problems in NP are harder than P-class problems and cannot be solved efficiently. This is considered one of the hardest unsolved problems in mathematics.'}, 'name': 'Respond', 'type': 'tool_use'}]
Tool Calls:
Respond (toolu_01ACZEPYEyqmpf3kA4VERXFY)
Call ID: toolu_01ACZEPYEyqmpf3kA4VERXFY
Args:
answer: With a Llama V3, the answer you'll see,
Whether P equals NP is a mystery!
The class P and NP, a puzzle so grand,
Solved or unsolved, the future's at hand.
The question remains, unanswered for now,
But with a Llama V3, we'll find out how!
reason: The question of whether P = NP is one of the most famous unsolved problems in computer science and mathematics. P and NP are complexity classes that describe how quickly problems can be solved by computers.
The P class contains problems that can be solved in polynomial time, meaning the time to solve the problem scales polynomially with the size of the input. The NP class contains problems where the solution can be verified in polynomial time, but there may not be a polynomial time algorithm to find the solution.
Whether P = NP is an open question - it is not known if every problem in NP can also be solved in polynomial time. If P = NP, it would mean that all problems with quickly verifiable solutions could also be quickly solved, which would have major implications for computing and cryptography. However, most experts believe that P ≠ NP, meaning some problems in NP are harder than P-class problems and cannot be solved efficiently. This is considered one of the hardest unsolved problems in mathematics.
在 [5]
已复制！
from typing import List, Optional
class OutputFormat(BaseModel):
sources: str = Field(
...,
description="The raw transcript / span you could cite to justify the choice.",
)
content: str = Field(..., description="The chosen value.")
class Moment(BaseModel):
quote: str = Field(..., description="The relevant quote from the transcript.")
description: str = Field(..., description="A description of the moment.")
expressed_preference: OutputFormat = Field(
..., description="The preference expressed in the moment."
)
class BackgroundInfo(BaseModel):
factoid: OutputFormat = Field(
..., description="Important factoid about the member."
)
professions: list
why: str = Field(..., description="Why this is important.")
class KeyMoments(BaseModel):
topic: str = Field(..., description="The topic of the key moments.")
happy_moments: List[Moment] = Field(
..., description="A list of key moments related to the topic."
)
tense_moments: List[Moment] = Field(
..., description="Moments where things were a bit tense."
)
sad_moments: List[Moment] = Field(
..., description="Moments where things where everyone was downtrodden."
)
background_info: list[BackgroundInfo]
moments_summary: str = Field(..., description="A summary of the key moments.")
class Member(BaseModel):
name: OutputFormat = Field(..., description="The name of the member.")
role: Optional[str] = Field(None, description="The role of the member.")
age: Optional[int] = Field(None, description="The age of the member.")
background_details: List[BackgroundInfo] = Field(
..., description="A list of background details about the member."
)
class InsightfulQuote(BaseModel):
quote: OutputFormat = Field(
..., description="An insightful quote from the transcript."
)
speaker: str = Field(..., description="The name of the speaker who said the quote.")
analysis: str = Field(
..., description="An analysis of the quote and its significance."
)
class TranscriptMetadata(BaseModel):
title: str = Field(..., description="The title of the transcript.")
location: OutputFormat = Field(
..., description="The location where the interview took place."
)
duration: str = Field(..., description="The duration of the interview.")
class TranscriptSummary(BaseModel):
metadata: TranscriptMetadata = Field(
..., description="Metadata about the transcript."
)
participants: List[Member] = Field(
..., description="A list of participants in the interview."
)
key_moments: List[KeyMoments] = Field(
..., description="A list of key moments from the interview."
)
insightful_quotes: List[InsightfulQuote] = Field(
..., description="A list of insightful quotes from the interview."
)
overall_summary: str = Field(
..., description="An overall summary of the interview."
)
next_steps: List[str] = Field(
..., description="A list of next steps or action items based on the interview."
)
other_stuff: List[OutputFormat]
from typing import List, Optional class OutputFormat(BaseModel): sources: str = Field( ..., description="您可以引用以证明选择的原始成绩单/跨度。", ) content: str = Field(..., description="选定的值。") class Moment(BaseModel): quote: str = Field(..., description="来自成绩单的相关引用。") description: str = Field(..., description="对时刻的描述。") expressed_preference: OutputFormat = Field( ..., description="在时刻中表达的偏好。" ) class BackgroundInfo(BaseModel): factoid: OutputFormat = Field( ..., description="关于成员的重要事实。") professions: list why: str = Field(..., description="为什么这很重要。") class KeyMoments(BaseModel): topic: str = Field(..., description="关键时刻的主题。") happy_moments: List[Moment] = Field( ..., description="与主题相关的关键时刻列表。" ) tense_moments: List[Moment] = Field( ..., description="事情有点紧张的时刻。" ) sad_moments: List[Moment] = Field( ..., description="每个人都沮丧的时刻。" ) background_info: list[BackgroundInfo] moments_summary: str = Field(..., description="关键时刻的摘要。") class Member(BaseModel): name: OutputFormat = Field(..., description="成员的姓名。") role: Optional[str] = Field(None, description="成员的角色。") age: Optional[int] = Field(None, description="成员的年龄。") background_details: List[BackgroundInfo] = Field( ..., description="关于成员的背景信息列表。" ) class InsightfulQuote(BaseModel): quote: OutputFormat = Field( ..., description="来自成绩单的深刻引用。") speaker: str = Field(..., description="说这句引语的人的名字。") analysis: str = Field( ..., description="对引语及其意义的分析。" ) class TranscriptMetadata(BaseModel): title: str = Field(..., description="成绩单的标题。") location: OutputFormat = Field( ..., description="采访发生的地点。") duration: str = Field(..., description="采访的时长。") class TranscriptSummary(BaseModel): metadata: TranscriptMetadata = Field( ..., description="关于成绩单的元数据。" ) participants: List[Member] = Field( ..., description="采访参与者列表。" ) key_moments: List[KeyMoments] = Field( ..., description="采访中的关键时刻列表。" ) insightful_quotes: List[InsightfulQuote] = Field( ..., description="来自采访的深刻引语列表。" ) overall_summary: str = Field( ..., description="对采访的总体概述。" ) next_steps: List[str] = Field( ..., description="根据采访得出的一系列下一步或行动项目。" ) other_stuff: List[OutputFormat]
让我们看看它在虚构的成绩单上的表现。
在 [6]
已复制！
transcript = [
(
"Pete",
"Hey Xu, Laura, thanks for hopping on this call. I've been itching to talk about this Drake and Kendrick situation.",
),
(
"Xu",
"No problem. As its my job, I've got some thoughts on this beef.",
),
(
"Laura",
"Yeah, I've got some insider info so this should be interesting.",
),
("Pete", "Dope. So, when do you think this whole thing started?"),
(
"Pete",
"Definitely was Kendrick's 'Control' verse that kicked it off.",
),
(
"Laura",
"Truth, but Drake never went after him directly. Just some subtle jabs here and there.",
),
(
"Xu",
"That's the thing with beefs like this, though. They've always been a a thing, pushing artists to step up their game.",
),
(
"Pete",
"For sure, and this beef has got the fans taking sides. Some are all about Drake's mainstream appeal, while others are digging Kendrick's lyrical skills.",
),
(
"Laura",
"I mean, Drake knows how to make a hit that gets everyone hyped. That's his thing.",
),
(
"Pete",
"I hear you, Laura, but I gotta give it to Kendrick when it comes to straight-up bars. The man's a beast on the mic.",
),
(
"Xu",
"It's wild how this beef is shaping fans.",
),
("Pete", "do you think these beefs can actually be good for hip-hop?"),
(
"Xu",
"Hell yeah, Pete. When it's done right, a beef can push the genre forward and make artists level up.",
),
("Laura", "eh"),
("Pete", "So, where do you see this beef going?"),
(
"Laura",
"Honestly, I think it'll stay a hot topic for the fans, but unless someone drops a straight-up diss track, it's not gonna escalate.",
),
("Laura", "ehhhhhh not sure"),
(
"Pete",
"I feel that. I just want both of them to keep dropping heat, beef or no beef.",
),
(
"Xu",
"I'm curious. May influence a lot of people. Make things more competitive. Bring on a whole new wave of lyricism.",
),
(
"Pete",
"Word. Hey, thanks for chopping it up with me, Xu and Laura. This was dope.",
),
("Xu", "Where are you going so fast?"),
(
"Laura",
"For real, I had a good time. Nice to get different perspectives on the situation.",
),
]
formatted = "\n".join(f"{x[0]}: {x[1]}" for x in transcript)
transcript = [ ( "Pete", "嘿，徐，劳拉，感谢你们加入这个电话会议。我一直很想聊聊德雷克和肯德里克的状况。", ), ( "Xu", "没问题。作为我的工作，我对这场争端有一些想法。", ), ( "Laura", "是的，我有一些内部消息，所以这应该很有趣。", ), ("Pete", "太棒了。那你认为这整件事是什么时候开始的？"), ( "Pete", "肯德里克的 'Control' 歌词绝对是导火索。", ), ( "Laura", "没错，但是德雷克从来没有直接攻击他。只是一些零星的微妙嘲讽。", ), ( "Xu", "不过，像这样的争端，它们一直都存在，推动着艺术家提高他们的水平。", ), ( "Pete", "当然，这场争端让粉丝们站队。有些人是德雷克主流吸引力的拥趸，而另一些人则沉醉于肯德里克的歌词技巧。", ), ( "Laura", "我的意思是，德雷克知道如何制作让所有人兴奋的热门歌曲。这是他的强项。", ), ( "Pete", "我明白，劳拉，但我必须承认，说到纯粹的歌词，肯德里克更胜一筹。他在麦克风前简直是野兽。", ), ( "Xu", "这场争端正在塑造粉丝们，这太疯狂了。", ), ("Pete", "你认为这些争端实际上对嘻哈文化有利吗？"), ( "Xu", "当然，皮特。如果做得好，争端可以推动这种类型的发展，让艺术家们提升水平。", ), ("Laura", "嗯"), ("Pete", "那么，你觉得这场争端会走向何方？"), ( "Laura", "说实话，我认为它会一直是粉丝们热议的话题，但除非有人直接发表一首diss歌曲，否则它不会升级。", ), ("Laura", "嗯嗯嗯 不确定"), ( "Pete", "我感觉到了。我只希望他们两个都能继续创作出优秀的音乐，无论有没有争端。", ), ( "Xu", "我很想知道。可能会影响很多人。让事情更有竞争性。带来一波全新的歌词风格。", ), ( "Pete", "没错。嘿，谢谢你们跟我聊了这么多，徐和劳拉。这次谈话很精彩。", ), ("Xu", "你这么急着去哪里？"), ( "Laura", "说真的，我很开心。很高兴能从不同的角度看待这场争端。", ), ] formatted = "\n".join(f"{x[0]}: {x[1]}" for x in transcript)
现在，运行我们的模型。我们**预期**GPT turbo 仍然无法处理这个具有挑战性的模板。
在 [7]
已复制！
tools = [TranscriptSummary]
bound_llm = bind_validator_with_retries(
llm,
tools=tools,
)
prompt = ChatPromptTemplate.from_messages(
[
("system", "Respond directly using the TranscriptSummary function."),
("placeholder", "{messages}"),
]
)
chain = prompt | bound_llm
try:
results = chain.invoke(
{
"messages": [
(
"user",
f"Extract the summary from the following conversation:\n\n<convo>\n{formatted}\n</convo>"
"\n\nRemember to respond using the TranscriptSummary function.",
)
]
},
)
results.pretty_print()
except ValueError as e:
print(repr(e))
tools = [TranscriptSummary] bound_llm = bind_validator_with_retries( llm, tools=tools, ) prompt = ChatPromptTemplate.from_messages( [ ("system", "直接使用 TranscriptSummary 函数进行回复。"), ("placeholder", "{messages}"), ] ) chain = prompt | bound_llm try: results = chain.invoke( { "messages": [ ( "user", f"从以下对话中提取摘要：\n\n\n{formatted}\n " "\n\n请记住使用 TranscriptSummary 函数进行回复。", ) ] }, ) results.pretty_print() except ValueError as e: print(repr(e))
ValueError('Could not extract a valid value in 3 attempts.')
JSONPatch¶
对于我们的简单案例，常规的重试方法效果很好，但它仍然无法在填充复杂模式时自我纠正。
LLM 最适合处理狭窄的任务。LLM 接口设计中一个行之有效的方法是简化每次 LLM 运行的任务。
一种方法是**修补**状态，而不是完全重新生成状态。一种方法是使用JSONPatch 操作。让我们试一试！
在下面，创建一个 JSONPatch 重试图。它的工作原理如下
- 第一次尝试：尝试生成完整输出。
- 重试：提示 LLM 在第一次输出的基础上生成**JSON 修补程序**来修复错误的生成。
回退 LLM 只需生成路径、操作（添加、删除、替换）和可选值的列表即可。由于 pydantic 验证错误在其错误中包含路径，因此 LLM 应该更加可靠。
在 [ ]
已复制！
%%capture --no-stderr
%pip install -U jsonpatch
%%capture --no-stderr %pip install -U jsonpatch
在 [12]
已复制！
import logging
logger = logging.getLogger("extraction")
def bind_validator_with_jsonpatch_retries(
llm: BaseChatModel,
*,
tools: list,
tool_choice: Optional[str] = None,
max_attempts: int = 3,
) -> Runnable[Union[List[AnyMessage], PromptValue], AIMessage]:
"""Binds validators + retry logic ensure validity of generated tool calls.
This method is similar to `bind_validator_with_retries`, but uses JSONPatch to correct
validation errors caused by passing in incorrect or incomplete parameters in a previous
tool call. This method requires the 'jsonpatch' library to be installed.
Using patch-based function healing can be more efficient than repopulating the entire
tool call from scratch, and it can be an easier task for the LLM to perform, since it typically
only requires a few small changes to the existing tool call.
Args:
llm (Runnable): The llm that will generate the initial messages (and optionally fallba)
tools (list): The tools to bind to the LLM.
tool_choice (Optional[str]): The tool choice to use.
max_attempts (int): The number of attempts to make.
Returns:
Runnable: A runnable that can be invoked with a list of messages and returns a single AI message.
"""
try:
import jsonpatch # type: ignore[import-untyped]
except ImportError:
raise ImportError(
"The 'jsonpatch' library is required for JSONPatch-based retries."
)
class JsonPatch(BaseModel):
"""A JSON Patch document represents an operation to be performed on a JSON document.
Note that the op and path are ALWAYS required. Value is required for ALL operations except 'remove'.
Examples:
```json
{"op": "add", "path": "/a/b/c", "patch_value": 1}
{"op": "replace", "path": "/a/b/c", "patch_value": 2}
{"op": "remove", "path": "/a/b/c"}
```
"""
op: Literal["add", "remove", "replace"] = Field(
...,
description="The operation to be performed. Must be one of 'add', 'remove', 'replace'.",
)
path: str = Field(
...,
description="A JSON Pointer path that references a location within the target document where the operation is performed.",
)
value: Any = Field(
...,
description="The value to be used within the operation. REQUIRED for 'add', 'replace', and 'test' operations.",
)
class PatchFunctionParameters(BaseModel):
"""Respond with all JSONPatch operation to correct validation errors caused by passing in incorrect or incomplete parameters in a previous tool call."""
tool_call_id: str = Field(
...,
description="The ID of the original tool call that generated the error. Must NOT be an ID of a PatchFunctionParameters tool call.",
)
reasoning: str = Field(
...,
description="Think step-by-step, listing each validation error and the"
" JSONPatch operation needed to correct it. "
"Cite the fields in the JSONSchema you referenced in developing this plan.",
)
patches: list[JsonPatch] = Field(
...,
description="A list of JSONPatch operations to be applied to the previous tool call's response.",
)
bound_llm = llm.bind_tools(tools, tool_choice=tool_choice)
fallback_llm = llm.bind_tools([PatchFunctionParameters])
def aggregate_messages(messages: Sequence[AnyMessage]) -> AIMessage:
# Get all the AI messages and apply json patches
resolved_tool_calls: Dict[Union[str, None], ToolCall] = {}
content: Union[str, List[Union[str, dict]]] = ""
for m in messages:
if m.type != "ai":
continue
if not content:
content = m.content
for tc in m.tool_calls:
if tc["name"] == PatchFunctionParameters.__name__:
tcid = tc["args"]["tool_call_id"]
if tcid not in resolved_tool_calls:
logger.debug(
f"JsonPatch tool call ID {tc['args']['tool_call_id']} not found."
f"Valid tool call IDs: {list(resolved_tool_calls.keys())}"
)
tcid = next(iter(resolved_tool_calls.keys()), None)
orig_tool_call = resolved_tool_calls[tcid]
current_args = orig_tool_call["args"]
patches = tc["args"].get("patches") or []
orig_tool_call["args"] = jsonpatch.apply_patch(
current_args,
patches,
)
orig_tool_call["id"] = tc["id"]
else:
resolved_tool_calls[tc["id"]] = tc.copy()
return AIMessage(
content=content,
tool_calls=list(resolved_tool_calls.values()),
)
def format_exception(error: BaseException, call: ToolCall, schema: Type[BaseModel]):
return (
f"Error:\n\n```\n{repr(error)}\n```\n"
"Expected Parameter Schema:\n\n" + f"```json\n{schema.schema_json()}\n```\n"
f"Please respond with a JSONPatch to correct the error for tool_call_id=[{call['id']}]."
)
validator = ValidationNode(
tools + [PatchFunctionParameters],
format_error=format_exception,
)
retry_strategy = RetryStrategy(
max_attempts=max_attempts,
fallback=fallback_llm,
aggregate_messages=aggregate_messages,
)
return _bind_validator_with_retries(
bound_llm,
validator=validator,
retry_strategy=retry_strategy,
tool_choice=tool_choice,
).with_config(metadata={"retry_strategy": "jsonpatch"})
import logging logger = logging.getLogger("extraction") def bind_validator_with_jsonpatch_retries( llm: BaseChatModel, *, tools: list, tool_choice: Optional[str] = None, max_attempts: int = 3, ) -> Runnable[Union[List[AnyMessage], PromptValue], AIMessage]: """绑定验证器 + 重试逻辑以确保生成的工具调用有效性。此方法类似于 `bind_validator_with_retries`，但使用 JSONPatch 来纠正由先前工具调用中传递错误或不完整参数引起的验证错误。 此方法需要安装 'jsonpatch' 库。 使用基于补丁的函数修复可能比从头重新填充整个工具调用更有效，而且对于 LLM 来说，这可能是一个更容易的任务，因为它通常只需要对现有的工具调用进行一些小的更改。 Args: llm (Runnable): 将生成初始消息（以及可选回退）的 llm tools (list): 要绑定到 LLM 的工具。 tool_choice (Optional[str]): 要使用的工具选择。 max_attempts (int): 要尝试的次数。 Returns: Runnable: 可以使用消息列表调用的可运行程序，并返回单个 AI 消息。 """ try: import jsonpatch # type: ignore[import-untyped] except ImportError: raise ImportError( "The 'jsonpatch' library is required for JSONPatch-based retries." ) class JsonPatch(BaseModel): """JSON Patch 文档表示要对 JSON 文档执行的操作。 请注意，op 和 path 始终是必需的。 除 'remove' 外，所有操作都需要 Value。 例子： ```json {"op": "add", "path": "/a/b/c", "patch_value": 1} {"op": "replace", "path": "/a/b/c", "patch_value": 2} {"op": "remove", "path": "/a/b/c"} ``` """ op: Literal["add", "remove", "replace"] = Field( ..., description="要执行的操作。必须是 'add'、'remove' 或 'replace' 之一。", ) path: str = Field( ..., description="JSON 指针路径，引用目标文档中执行操作的位置。", ) value: Any = Field( ..., description="要在操作中使用的值。'add'、'replace' 和 'test' 操作都需要。", ) class PatchFunctionParameters(BaseModel): """使用所有 JSONPatch 操作来纠正由先前工具调用中传递错误或不完整参数引起的验证错误。""" tool_call_id: str = Field( ..., description="生成错误的原始工具调用的 ID。 绝不能是 PatchFunctionParameters 工具调用的 ID。", ) reasoning: str = Field( ..., description="逐步思考，列出每个验证错误以及纠正它的 JSONPatch 操作。 " "引用你在开发此计划时参考的 JSONSchema 中的字段。", ) patches: list[JsonPatch] = Field( ..., description="要应用于先前工具调用响应的 JSONPatch 操作列表。", ) bound_llm = llm.bind_tools(tools, tool_choice=tool_choice) fallback_llm = llm.bind_tools([PatchFunctionParameters]) def aggregate_messages(messages: Sequence[AnyMessage]) -> AIMessage: # 获取所有 AI 消息并应用 json 补丁 resolved_tool_calls: Dict[Union[str, None], ToolCall] = {} content: Union[str, List[Union[str, dict]]] = "" for m in messages: if m.type != "ai": continue if not content: content = m.content for tc in m.tool_calls: if tc["name"] == PatchFunctionParameters.__name__: tcid = tc["args"]["tool_call_id"] if tcid not in resolved_tool_calls: logger.debug( f"JsonPatch 工具调用 ID {tc['args']['tool_call_id']} 未找到。" f"有效的工具调用 ID：{list(resolved_tool_calls.keys())}" ) tcid = next(iter(resolved_tool_calls.keys()), None) orig_tool_call = resolved_tool_calls[tcid] current_args = orig_tool_call["args"] patches = tc["args"].get("patches") or [] orig_tool_call["args"] = jsonpatch.apply_patch( current_args, patches, ) orig_tool_call["id"] = tc["id"] else: resolved_tool_calls[tc["id"]] = tc.copy() return AIMessage( content=content, tool_calls=list(resolved_tool_calls.values()), ) def format_exception(error: BaseException, call: ToolCall, schema: Type[BaseModel]): return ( f"Error:\n\n```\n{repr(error)}\n```\n" "Expected Parameter Schema:\n\n" + f"```json\n{schema.schema_json()}\n```\n" f"请使用 JSONPatch 响应以纠正 tool_call_id=[{call['id']}] 的错误。" ) validator = ValidationNode( tools + [PatchFunctionParameters], format_error=format_exception, ) retry_strategy = RetryStrategy( max_attempts=max_attempts, fallback=fallback_llm, aggregate_messages=aggregate_messages, ) return _bind_validator_with_retries( bound_llm, validator=validator, retry_strategy=retry_strategy, tool_choice=tool_choice, ).with_config(metadata={"retry_strategy": "jsonpatch"})
在 [13]
已复制！
bound_llm = bind_validator_with_jsonpatch_retries(llm, tools=tools)
bound_llm = bind_validator_with_jsonpatch_retries(llm, tools=tools)
在 [14]
已复制！
from IPython.display import Image, display
try:
display(Image(bound_llm.get_graph().draw_mermaid_png()))
except Exception:
pass
from IPython.display import Image, display try: display(Image(bound_llm.get_graph().draw_mermaid_png())) except Exception: pass
在 [15]
已复制！
chain = prompt | bound_llm
results = chain.invoke(
{
"messages": [
(
"user",
f"Extract the summary from the following conversation:\n\n<convo>\n{formatted}\n</convo>",
),
]
},
)
results.pretty_print()
chain = prompt | bound_llm results = chain.invoke( { "messages": [ ( "user", f"从以下对话中提取摘要：\n\n\n{formatted}\n ", ), ] }, ) results.pretty_print()
================================== Ai Message ==================================
[{'text': 'Here is a summary of the key points from the conversation:', 'type': 'text'}, {'id': 'toolu_01JjnQVgzPKLCJxXgEppQpfD', 'input': {'key_moments': [{'topic': 'Drake and Kendrick Lamar beef', 'happy_moments': [{'quote': "It's wild how this beef is shaping fans.", 'description': 'The beef is generating a lot of interest and debate among fans.', 'expressed_preference': {'content': 'The beef can push the genre forward and make artists level up.', 'sources': "When it's done right, a beef can push the genre forward and make artists level up."}}, {'quote': 'I just want both of them to keep dropping heat, beef or no beef.', 'description': 'The key is for Drake and Kendrick to keep making great music regardless of their beef.', 'expressed_preference': {'content': 'Wants Drake and Kendrick to keep making great music, beef or no beef.', 'sources': 'I just want both of them to keep dropping heat, beef or no beef.'}}], 'tense_moments': [{'quote': 'Eh', 'description': 'Unclear if the beef is good for hip-hop.', 'expressed_preference': {'content': 'Unsure if the beef is good for hip-hop.', 'sources': 'Eh'}}], 'sad_moments': [{'quote': "Honestly, I think it'll stay a hot topic for the fans, but unless someone drops a straight-up diss track, it's not gonna escalate.", 'description': "The beef may just stay a topic of discussion among fans, but likely won't escalate unless they release direct diss tracks.", 'expressed_preference': {'content': "The beef will likely remain a topic of discussion but won't escalate unless they release diss tracks.", 'sources': "Honestly, I think it'll stay a hot topic for the fans, but unless someone drops a straight-up diss track, it's not gonna escalate."}}], 'background_info': [{'factoid': {'content': "Kendrick's 'Control' verse kicked off the beef.", 'sources': "Definitely was Kendrick's 'Control' verse that kicked it off."}, 'professions': [], 'why': 'This was the event that started the back-and-forth between Drake and Kendrick.'}, {'factoid': {'content': 'Drake never went directly after Kendrick, just some subtle jabs.', 'sources': 'Drake never went after him directly. Just some subtle jabs here and there.'}, 'professions': [], 'why': "Describes the nature of Drake's response to Kendrick's 'Control' verse."}], 'moments_summary': "The conversation covers the ongoing beef between Drake and Kendrick Lamar, including how it started with Kendrick's 'Control' verse, the subtle jabs back and forth, and debate over whether the beef is ultimately good for hip-hop. There are differing views on whether it will escalate beyond just being a topic of discussion among fans."}]}, 'name': 'TranscriptSummary', 'type': 'tool_use'}]
Tool Calls:
TranscriptSummary (toolu_017FF4ZMezU4sv87aa8cLjRT)
Call ID: toolu_017FF4ZMezU4sv87aa8cLjRT
Args:
key_moments: [{'topic': 'Drake and Kendrick Lamar beef', 'happy_moments': [{'quote': "It's wild how this beef is shaping fans.", 'description': 'The beef is generating a lot of interest and debate among fans.', 'expressed_preference': {'content': 'The beef can push the genre forward and make artists level up.', 'sources': "When it's done right, a beef can push the genre forward and make artists level up."}}, {'quote': 'I just want both of them to keep dropping heat, beef or no beef.', 'description': 'The key is for Drake and Kendrick to keep making great music regardless of their beef.', 'expressed_preference': {'content': 'Wants Drake and Kendrick to keep making great music, beef or no beef.', 'sources': 'I just want both of them to keep dropping heat, beef or no beef.'}}], 'tense_moments': [{'quote': 'Eh', 'description': 'Unclear if the beef is good for hip-hop.', 'expressed_preference': {'content': 'Unsure if the beef is good for hip-hop.', 'sources': 'Eh'}}], 'sad_moments': [{'quote': "Honestly, I think it'll stay a hot topic for the fans, but unless someone drops a straight-up diss track, it's not gonna escalate.", 'description': "The beef may just stay a topic of discussion among fans, but likely won't escalate unless they release direct diss tracks.", 'expressed_preference': {'content': "The beef will likely remain a topic of discussion but won't escalate unless they release diss tracks.", 'sources': "Honestly, I think it'll stay a hot topic for the fans, but unless someone drops a straight-up diss track, it's not gonna escalate."}}], 'background_info': [{'factoid': {'content': "Kendrick's 'Control' verse kicked off the beef.", 'sources': "Definitely was Kendrick's 'Control' verse that kicked it off."}, 'professions': [], 'why': 'This was the event that started the back-and-forth between Drake and Kendrick.'}, {'factoid': {'content': 'Drake never went directly after Kendrick, just some subtle jabs.', 'sources': 'Drake never went after him directly. Just some subtle jabs here and there.'}, 'professions': [], 'why': "Describes the nature of Drake's response to Kendrick's 'Control' verse."}], 'moments_summary': "The conversation covers the ongoing beef between Drake and Kendrick Lamar, including how it started with Kendrick's 'Control' verse, the subtle jabs back and forth, and debate over whether the beef is ultimately good for hip-hop. There are differing views on whether it will escalate beyond just being a topic of discussion among fans."}]
metadata: {'title': 'Drake and Kendrick Beef', 'location': {'sources': 'Conversation transcript', 'content': 'Teleconference'}, 'duration': '25 minutes'}
participants: [{'name': {'sources': 'Conversation transcript', 'content': 'Pete'}, 'background_details': []}, {'name': {'sources': 'Conversation transcript', 'content': 'Xu'}, 'background_details': []}, {'name': {'sources': 'Conversation transcript', 'content': 'Laura'}, 'background_details': []}]
insightful_quotes: []
overall_summary:
next_steps: []
other_stuff: []
它有效！¶
重试是减少函数调用失败的简单方法。虽然随着更强大的 LLM 的出现，重试可能变得不再必要，但数据验证对于控制 LLM 与软件堆栈其余部分的交互至关重要。
如果你注意到高重试率（使用 LangSmith 等可观察性工具），你可以设置一个规则，将故障案例与更正后的值一起发送到数据集，然后自动将这些案例编程到你的提示或模式中（或使用它们作为少量示例来获得语义相关演示）。