LangChain#

1
from pydantic import BaseModel, Field
2
from langchain_openai import ChatOpenAI
3

4
class UserProfile(BaseModel):
5
    """用户档案数据模型"""
6
    name: str = Field(description="用户姓名")
7
    age: int = Field(description="用户年龄", ge=0, le=150)
8
    email: str = Field(description="邮箱地址")
9
    skills: List[str] = Field(description="技能列表")
10

11
# 创建结构化输出模型
12
model = ChatOpenAI(model="gpt-4o-mini")
13
structured_llm = model.with_structured_output(UserProfile)
14

15
# 使用结构化输出
16
result = structured_llm.invoke("提取用户信息：张三，25岁，邮箱zhang@example.com，擅长Python和数据分析")
17
# result 是 UserProfile 类型的对象
18
print(f"姓名: {result.name}, 年龄: {result.age}")

1
class Address(BaseModel):
2
    """地址信息"""
3
    street: str = Field(description="街道地址")
4
    city: str = Field(description="城市")
5
    country: str = Field(description="国家")
6

7
class Company(BaseModel):
8
    """公司信息"""
9
    name: str = Field(description="公司名称")
10
    address: Address = Field(description="公司地址")
11
    employees: List[UserProfile] = Field(description="员工列表")
12

13
# 处理复杂嵌套结构
14
company_llm = model.with_structured_output(Company)

1
from enum import Enum
2

3
class TaskStatus(str, Enum):
4
    """任务状态枚举"""
5
    PENDING = "pending"
6
    IN_PROGRESS = "in_progress"
7
    COMPLETED = "completed"
8
    CANCELLED = "cancelled"
9

10
class Task(BaseModel):
11
    """任务信息"""
12
    title: str = Field(description="任务标题")
13
    status: TaskStatus = Field(description="任务状态")
14
    priority: int = Field(description="优先级", ge=1, le=5)
15

16
task_llm = model.with_structured_output(Task)

1
# 包含动态字段的模型需要使用 function_calling
2
class FlexibleResponse(BaseModel):
3
    core_data: str = Field(description="核心数据")
4
    metadata: Dict[str, Any] = Field(description="动态元数据")
5

6
# 必须指定 method="function_calling"
7
flexible_llm = model.with_structured_output(
8
    FlexibleResponse,
9
    method="function_calling"
10
)

1
from langchain_core.output_parsers import PydanticOutputParser
2

3
# 创建解析器
4
parser = PydanticOutputParser(pydantic_object=UserProfile)
5

6
# 获取格式指令
7
format_instructions = parser.get_format_instructions()
8

9
# 构建提示模板
10
from langchain_core.prompts import ChatPromptTemplate
11

12
prompt = ChatPromptTemplate.from_template(
13
    "提取用户信息：{query}\n{format_instructions}"
14
).partial(format_instructions=format_instructions)
15

16
# 创建完整链
17
chain = prompt | model | parser
18
result = chain.invoke({"query": "张三，工程师，28岁"})
19
# result 是 UserProfile 对象

1
# 流式JSON解析
2
from langchain_core.output_parsers import SimpleJsonOutputParser
3

4
streaming_parser = SimpleJsonOutputParser()
5

6
# 流式处理
7
for chunk in (model | streaming_parser).stream("生成用户数据"):
8
    print(chunk, end="", flush=True)

1
from langchain_core.output_parsers import OutputFixingParser
2

3
# 基础解析器
4
base_parser = JsonOutputParser()
5

6
# 包装为自动修复解析器
7
fixing_parser = OutputFixingParser.from_llm(
8
    parser=base_parser,
9
    llm=ChatOpenAI(temperature=0.1)  # 使用低温度提高修复准确性
10
)
11

12
# 自动处理格式错误
13
broken_json = '{"name": "张三", "age": 25'  # 缺少闭合括号
14
fixed_result = fixing_parser.parse(broken_json)  # 自动修复成功

1
from langchain_core.output_parsers import RetryWithErrorOutputParser
2

3
# 创建重试解析器
4
retry_parser = RetryWithErrorOutputParser.from_llm(
5
    parser=PydanticOutputParser(pydantic_object=UserProfile),
6
    llm=ChatOpenAI(temperature=0.1),
7
    max_retries=3  # 最多重试3次
8
)
9

10
# 自动处理解析失败并重试
11
chain = prompt | model | retry_parser
12
result = chain.invoke({"query": "用户信息提取"})

1
# 构建三层错误处理
2
base_parser = PydanticOutputParser(pydantic_object=ComplexModel)
3

4
# 第一层：格式修复
5
fixing_parser = OutputFixingParser.from_llm(
6
    parser=base_parser,
7
    llm=ChatOpenAI(temperature=0.1)
8
)
9

10
# 第二层：重试机制
11
ultimate_parser = RetryWithErrorOutputParser.from_llm(
12
    parser=fixing_parser,
13
    llm=ChatOpenAI(temperature=0.1),
14
    max_retries=2
15
)
16

17
ultimate_chain = prompt | model | ultimate_parser

1
from langchain_core.output_parsers import BaseOutputParser
2
from typing import List
3

4
class ListOutputParser(BaseOutputParser[List[str]]):
5
    """列表解析器 - 分隔符解析"""
6

7
    def __init__(self, separator: str = ","):
8
        self.separator = separator
9

10
    def parse(self, text: str) -> List[str]:
11
        """解析分隔符分隔的列表"""
12
        return [item.strip() for item in text.split(self.separator)]
13

14
    def get_format_instructions(self) -> str:
15
        """返回格式指令"""
16
        return f"请用{self.separator}分隔列表项目"
17

18
# 使用自定义解析器
19
list_parser = ListOutputParser(separator="|")
20
list_chain = model | list_parser

1
class ConditionalOutputParser(BaseOutputParser):
2
    """条件解析器 - 智能格式识别"""
3

4
    def parse(self, text: str) -> Union[str, Dict, List]:
5
        """根据内容特征选择解析策略"""
6
        text = text.strip()
7

8
        if text.startswith('{') and text.endswith('}'):
9
            # JSON对象格式
10
            return json.loads(text)
11
        elif text.startswith('[') and text.endswith(']'):
12
            # JSON数组格式
13
            return json.loads(text)
14
        elif ',' in text:
15
            # 逗号分隔的列表
16
            return [item.strip() for item in text.split(',')]
17
        else:
18
            # 纯文本
19
            return text
20

21
    def get_format_instructions(self) -> str:
22
        return "可以返回JSON、列表或纯文本格式"

1
class ChainedOutputParser(BaseOutputParser[Dict[str, Any]]):
2
    """链式解析器 - 多策略组合"""
3

4
    def __init__(self, parsers: List[Tuple[str, BaseOutputParser]]):
5
        self.parsers = parsers
6

7
    def parse(self, text: str) -> Dict[str, Any]:
8
        """依次应用多个解析器"""
9
        results = {"original": text}
10

11
        for name, parser in self.parsers:
12
            try:
13
                results[name] = parser.parse(text)
14
            except Exception as e:
15
                results[f"{name}_error"] = str(e)
16

17
        return results
18

19
# 创建链式解析器
20
chained_parser = ChainedOutputParser([
21
    ("json", JsonOutputParser()),
22
    ("list", ListOutputParser()),
23
    ("yaml", YamlOutputParser())
24
])

1
class ValidatingOutputParser(BaseOutputParser[Dict]):
2
    """验证解析器 - 数据质量保证"""
3

4
    def __init__(self, base_parser: BaseOutputParser, validators: List[callable]):
5
        self.base_parser = base_parser
6
        self.validators = validators
7

8
    def parse(self, text: str) -> Dict:
9
        """解析并验证数据"""
10
        result = self.base_parser.parse(text)
11

12
        # 应用验证规则
13
        for validator in self.validators:
14
            result = validator(result)
15

16
        return result
17

18
def validate_user_age(data: Dict) -> Dict:
19
    """验证用户年龄合理性"""
20
    if 'age' in data and not (0 <= data['age'] <= 150):
21
        data['age'] = max(0, min(data['age'], 150))
22
    return data
23

24
# 使用验证解析器
25
validating_parser = ValidatingOutputParser(
26
    base_parser=JsonOutputParser(),
27
    validators=[validate_user_age]
28
)

1
需要结构化输出？
2
├─ 是 → 数据结构复杂（包含动态字段）？
3
│   ├─ 是 → 使用 with_structured_output(method="function_calling")
4
│   └─ 否 → 使用 with_structured_output() (默认模式)
5
└─ 否 → 需要格式转换？
6
    ├─ 是 → 使用合适的OutputParser (Json/XML/YAML)
7
    └─ 否 → 使用 StrOutputParser

1
def robust_structured_output(query: str, model_class: BaseModel, fallback_parser=None):
2
    """鲁棒的结构化输出处理"""
3
    try:
4
        # 主要方法：结构化输出
5
        structured_llm = model.with_structured_output(model_class)
6
        return structured_llm.invoke(query)
7
    except Exception as e:
8
        if fallback_parser:
9
            # 降级方法：使用解析器
10
            return fallback_parser.parse(model.invoke(query).content)
11
        else:
12
            raise e

1
import time
2
from typing import Any
3

4
def monitored_parse(parser: BaseOutputParser, text: str) -> Dict[str, Any]:
5
    """监控解析性能"""
6
    start_time = time.time()
7
    try:
8
        result = parser.parse(text)
9
        success = True
10
        error = None
11
    except Exception as e:
12
        result = None
13
        success = False
14
        error = str(e)
15

16
    end_time = time.time()
17

18
    return {
19
        "result": result,
20
        "success": success,
21
        "error": error,
22
        "parse_time": end_time - start_time,
23
        "input_length": len(text)
24
    }

1
def enhance_format_instructions(base_instructions: str, examples: List[str] = None) -> str:
2
    """增强格式指令"""
3
    enhanced = base_instructions
4

5
    if examples:
6
        enhanced += "\n\n示例格式:\n"
7
        for i, example in enumerate(examples, 1):
8
            enhanced += f"{i}. {example}\n"
9

10
    enhanced += "\n注意：严格按照上述格式返回，确保数据完整性。"
11
    return enhanced

1
from typing import TypeVar, Generic
2

3
T = TypeVar('T')
4

5
class TypedOutputParser(BaseOutputParser[T], Generic[T]):
6
    """类型安全的解析器基类"""
7

8
    def __init__(self, target_type: Type[T]):
9
        self.target_type = target_type
10

11
    def parse(self, text: str) -> T:
12
        # 实现类型安全的解析逻辑
13
        pass