Source code for langchain.output_parsers.structured

from __future__ import annotations

from typing import Any, List

from langchain_core.output_parsers import BaseOutputParser
from langchain_core.output_parsers.json import parse_and_check_json_markdown
from langchain_core.pydantic_v1 import BaseModel

from langchain.output_parsers.format_instructions import (
    STRUCTURED_FORMAT_INSTRUCTIONS,
    STRUCTURED_FORMAT_SIMPLE_INSTRUCTIONS,
)

line_template = '\t"{name}": {type}  // {description}'


[docs]class ResponseSchema(BaseModel): """用于结构化输出解析器响应的模式。""" name: str """模式的名称。""" description: str """模式的描述。""" type: str = "string" """响应的类型。"""
def _get_sub_string(schema: ResponseSchema) -> str: return line_template.format( name=schema.name, description=schema.description, type=schema.type )
[docs]class StructuredOutputParser(BaseOutputParser): """解析LLM调用的输出,转换为结构化输出。""" response_schemas: List[ResponseSchema] """响应的模式。"""
[docs] @classmethod def from_response_schemas( cls, response_schemas: List[ResponseSchema] ) -> StructuredOutputParser: return cls(response_schemas=response_schemas)
[docs] def get_format_instructions(self, only_json: bool = False) -> str: """获取输出解析器的格式说明。 示例: ```python from langchain.output_parsers.structured import ( StructuredOutputParser, ResponseSchema ) response_schemas = [ ResponseSchema( name="foo", description="a list of strings", type="List[string]" ), ResponseSchema( name="bar", description="a string", type="string" ), ] parser = StructuredOutputParser.from_response_schemas(response_schemas) print(parser.get_format_instructions()) # noqa: T201 输出: # 输出应该是一个Markdown代码片段,格式如下 # 包括前导和尾随的 "```json" 和 "```": # # ```json # { # "foo": List[string] // a list of strings # "bar": string // a string # } # ``` 参数: only_json (bool): 如果为True,则只返回Markdown代码片段中的json部分,不包括引导文本。默认为False。 """ schema_str = "\n".join( [_get_sub_string(schema) for schema in self.response_schemas] ) if only_json: return STRUCTURED_FORMAT_SIMPLE_INSTRUCTIONS.format(format=schema_str) else: return STRUCTURED_FORMAT_INSTRUCTIONS.format(format=schema_str)
[docs] def parse(self, text: str) -> Any: expected_keys = [rs.name for rs in self.response_schemas] return parse_and_check_json_markdown(text, expected_keys)
@property def _type(self) -> str: return "structured"