使用Anthropic API生成结构化输出的工具

Anthropic API最近增加了工具使用功能。

这对于生成结构化输出非常有用。

! pip install -U langchain-anthropic

可选配置：

import os

os.environ['LANGCHAIN_TRACING_V2'] = 'true'  # 启用追踪
os.environ['LANGCHAIN_API_KEY'] =  # 设置你的API密钥

我们如何使用工具来产生结构化输出？

函数调用/工具使用仅生成有效载荷。

有效载荷通常是JSON字符串，可以传递给API，或者在本例中，传递给解析器以产生结构化输出。

LangChain提供了llm.with_structured_output(schema)，使得生成符合模式的结构化输出变得非常容易。

from langchain_anthropic import ChatAnthropic
from langchain_core.prompts import ChatPromptTemplate
from langchain_core.pydantic_v1 import BaseModel, Field

# 数据模型
class Code(BaseModel):
    """代码输出"""

# LLM
llm = ChatAnthropic(
    model="claude-3-opus-20240229",
    default_headers={"anthropic-beta": "tools-2024-04-04"},
)

# 结构化输出，包括原始内容将捕获原始输出和解析器错误
structured_llm = llm.with_structured_output(Code, include_raw=True)
code_output = structured_llm.invoke(
    "Write a python program that prints the string 'hello world' and tell me how it works in a sentence"
)

初始推理阶段：

code_output["raw"].content[0]

工具调用：

code_output["raw"].content[1]

JSON字符串：

code_output["raw"].content[1]["input"]

错误：

error = code_output["parsing_error"]
error

结果：

parsed_result = code_output["parsed"]

Python：

parsed_result.prefix

Python：

parsed_result.imports

Python：

parsed_result.code

更具挑战性的例子：

动机例子，用于工具使用/结构化输出。

这里是我们想要回答有关代码问题的一些文档。

from bs4 import BeautifulSoup as Soup
from langchain_community.document_loaders.recursive_url_loader import RecursiveUrlLoader

LCEL文档

url = "https://python.langchain.com/docs/expression_language/"
loader = RecursiveUrlLoader(
    url=url, max_depth=20, extractor=lambda x: Soup(x, "html.parser").text
)
docs = loader.load()

根据URL对列表进行排序并获取文本

d_sorted = sorted(docs, key=lambda x: x.metadata["source"])
d_reversed = list(reversed(d_sorted))
concatenated_content = "\n\n\n --- \n\n\n".join(
    [doc.page_content for doc in d_reversed]
)

问题：

如果我们想强制使用工具怎么办？

我们可以使用回退策略。

让我们选择一个代码生成提示，根据我的一些测试，它没有正确调用工具。

我们看看是否可以纠正这个问题。

# 此代码生成提示调用工具使用
code_gen_prompt_working = ChatPromptTemplate.from_messages(
    [
        (
            "system",
            """ You are a coding assistant with expertise in LCEL, LangChain expression language.
Here is the LCEL documentation:
 -------
{context}
 -------
Answer the user question based on the
above provided documentation. Ensure any code you provide can be executed with all required imports and variables
defined. Structure your answer: 1) a prefix describing the code solution, 2) the imports, 3) the functioning code block.
Invoke the code tool to structure the output correctly.

Here is the user question:""",
        ),
        ("placeholder", "{messages}"),
    ]
)

# 此代码生成提示没有调用工具使用
code_gen_prompt_bad = ChatPromptTemplate.from_messages(
    [
        (
            "system",
            """You are a coding assistant with expertise in LCEL, LangChain expression language.
Here is a full set of LCEL documentation:
 -------
{context}
 -------
Answer the user question based on the above provided documentation. Ensure any code you provide can be executed
with all required imports and variables defined. Structure your answer with a description of the code solution.

Then list the imports. And finally list the functioning code block. Here is the user question:""",
        ),
        ("placeholder", "{messages}"),
    ]
)

数据模型：

class Code(BaseModel):
    """代码输出"""

LLM：

llm = ChatAnthropic(
    model="claude-3-opus-20240229",
    default_headers={"anthropic-beta": "tools-2024-04-04"},
)

结构化输出：

包括原始内容将捕获原始输出和解析器错误

structured_llm = llm.with_structured_output(Code, include_raw=True)

检查错误：

def check_claude_output(tool_output):
    """检查解析错误或未能调用工具"""

带有输出检查的链：

code_chain = code_gen_prompt_bad | structured_llm | check_claude_output

让我们添加检查并重试。

def insert_errors(inputs):
    """在消息中插入错误"""

这将作为回退链运行。

fallback_chain = insert_errors | code_chain
N = 3  # 最大重试次数
code_chain_re_try = code_chain.with_fallbacks(
    fallbacks=[fallback_chain] * N, exception_key="error"
)

测试：

messages = [("user", "How do I build a RAG chain in LCEL?")]
code_output_lcel = code_chain_re_try.invoke(
    {"context": concatenated_content, "messages": messages}
)

Python：

parsed_result_lcel = code_output_lcel["parsed"]

Python：

parsed_result_lcel.prefix

Python：

parsed_result_lcel.imports

Python：

parsed_result_lcel.code

示例跟踪捕获错误并纠正：

跟踪链接

---

介绍了如何使用Anthropic API来生成结构化输出。它首先建议安装langchain-anthropic库，然后通过设置环境变量启用追踪和API密钥。文件中展示了如何定义数据模型，创建ChatAnthropic实例，并使用with_structured_output方法来生成符合特定模式的结构化输出。接着，通过示例代码演示了如何调用API、处理原始输出、捕获解析错误，并展示了如何通过回退机制和错误检查来增强工具调用的鲁棒性。最后，提供了一个测试用例，展示了如何使用构建的链来回答问题并获取结构化代码输出。