模块 json_schema

is_valid_json

def is_valid_json(s: str) -> bool

检查提供的字符串是否是有效的 JSON。

参数:

s：要检查的字符串。

返回值:

True 如果字符串是有效的 JSON；否则，False.

JsonSchemaValidator

验证的消息的 JSON 内容ChatMessage 根据指定的 JSON Schema。

如果消息的 JSON 内容符合提供的模式，则消息将通过“validated”输出。如果 JSON 内容不符合模式，则消息将通过“validation_error”输出。在后一种情况下，错误消息将使用提供的error_template 或默认模板。这些错误 ChatMessages 可以被 Haystack 2.x 恢复循环中的 LLM 使用。

使用示例

from haystack import Pipeline
from haystack.components.generators.chat import OpenAIChatGenerator
from haystack.components.joiners import BranchJoiner
from haystack.components.validators import JsonSchemaValidator
from haystack import component
from haystack.dataclasses import ChatMessage


@component
class MessageProducer:

    @component.output_types(messages=list[ChatMessage])
    def run(self, messages: list[ChatMessage]) -> dict:
        return {"messages": messages}


p = Pipeline()
p.add_component("llm", OpenAIChatGenerator(model="gpt-4-1106-preview",
                                           generation_kwargs={"response_format": {"type": "json_object"}}))
p.add_component("schema_validator", JsonSchemaValidator())
p.add_component("joiner_for_llm", BranchJoiner(list[ChatMessage]))
p.add_component("message_producer", MessageProducer())

p.connect("message_producer.messages", "joiner_for_llm")
p.connect("joiner_for_llm", "llm")
p.connect("llm.replies", "schema_validator.messages")
p.connect("schema_validator.validation_error", "joiner_for_llm")

result = p.run(data={
    "message_producer": {
        "messages":[ChatMessage.from_user("Generate JSON for person with name 'John' and age 30")]},
        "schema_validator": {
            "json_schema": {
                "type": "object",
                "properties": {"name": {"type": "string"},
                "age": {"type": "integer"}
            }
        }
    }
})
print(result)
>> {'schema_validator': {'validated': [ChatMessage(_role=<ChatRole.ASSISTANT: 'assistant'>,
_content=[TextContent(text="\n{\n  "name": "John",\n  "age": 30\n}")],
_name=None, _meta={'model': 'gpt-4-1106-preview', 'index': 0,
'finish_reason': 'stop', 'usage': {'completion_tokens': 17, 'prompt_tokens': 20, 'total_tokens': 37}})]}}

JsonSchemaValidator.init

def __init__(json_schema: Optional[dict[str, Any]] = None,
             error_template: Optional[str] = None)

初始化 JsonSchemaValidator 组件。

参数:

json_schema：一个代表 JSON schema 的字典，消息的内容将根据该模式进行验证。
error_template：一个自定义模板字符串，用于在验证失败时格式化错误消息。

JsonSchemaValidator.run

@component.output_types(validated=list[ChatMessage],
                        validation_error=list[ChatMessage])
def run(messages: list[ChatMessage],
        json_schema: Optional[dict[str, Any]] = None,
        error_template: Optional[str] = None) -> dict[str, list[ChatMessage]]

根据指定的 json schema 验证提供的消息中的最后一条。

如果验证成功，则消息通过“validated”输出。如果验证失败，则消息通过“validation_error”输出。

参数:

messages：要验证的 ChatMessage 实例列表。此列表中的最后一条消息是要验证的消息。
json_schema：一个代表 JSON schema 的字典，消息的内容将根据该模式进行验证。如果未提供，则使用组件 init 中的模式。
error_template：一个自定义模板字符串，用于在验证时格式化错误消息。如果未提供，则使用组件 init 中的error_template。

引发:

ValueError：如果没有提供 JSON schema，或者消息内容不是字典或字典列表。