Structured Output va Parsing

1from pydantic import BaseModel, Field
2from typing import List, Optional
3from langchain_openai import ChatOpenAI
4
5class ProductReview(BaseModel):
6    product_name: str = Field(description="Ten san pham")
7    rating: int = Field(description="Rating 1-5", ge=1, le=5)
8    pros: List[str] = Field(description="Diem manh")
9    cons: List[str] = Field(description="Diem yeu")
10    summary: str = Field(description="Tom tat ngan")
11    recommend: bool = Field(description="Co recommend khong")
12
13llm = ChatOpenAI(model="gpt-4o-mini", temperature=0)
14structured_llm = llm.with_structured_output(ProductReview)
15
16review_text = """
17Da dung iPhone 15 Pro duoc 3 thang. Camera chup rat dep, 
18chip A17 nhanh. Pin kha tot, dung ca ngay. Nhung gia qua cao 
19va nang hon iPhone 14. Nhin chung hai long, se gioi thieu 
20cho ban be.
21"""
22
23result = structured_llm.invoke(
24    f"Phan tich review sau:\n{review_text}"
25)
26print(f"Product: {result.product_name}")
27print(f"Rating: {result.rating}/5")
28print(f"Recommend: {result.recommend}")

1class Address(BaseModel):
2    street: str
3    city: str
4    country: str
5
6class ContactInfo(BaseModel):
7    name: str
8    email: Optional[str] = None
9    phone: Optional[str] = None
10    address: Optional[Address] = None
11
12class DocumentExtraction(BaseModel):
13    contacts: List[ContactInfo]
14    dates: List[str]
15    amounts: List[str]
16    action_items: List[str]
17
18extract_llm = llm.with_structured_output(DocumentExtraction)
19result = extract_llm.invoke(f"Extract info tu:\n{document_text}")

1from langchain_core.output_parsers import JsonOutputParser
2
3parser = JsonOutputParser()
4
5template = ChatPromptTemplate.from_messages([
6    ("system", """Phan tich text va tra ve JSON voi format:
7    {{
8        "entities": ["list of named entities"],
9        "topics": ["main topics"],
10        "language": "detected language",
11        "word_count": number
12    }}"""),
13    ("human", "{text}")
14])
15
16chain = template | llm | parser
17result = chain.invoke({"text": "Your text here..."})
18# result la Python dict

1from langchain_core.output_parsers import BaseOutputParser
2import re
3
4class ScoreParser(BaseOutputParser):
5    def parse(self, text: str) -> dict:
6        scores = {}
7        patterns = {
8            "relevance": r"Relevance:\s*(\d+)/10",
9            "clarity": r"Clarity:\s*(\d+)/10",
10            "accuracy": r"Accuracy:\s*(\d+)/10"
11        }
12        for key, pattern in patterns.items():
13            match = re.search(pattern, text)
14            scores[key] = int(match.group(1)) if match else 0
15        return scores
16
17score_chain = (
18    ChatPromptTemplate.from_messages([
19        ("system", """Danh gia text theo 3 tieu chi, moi tieu chi 1-10:
20        Relevance: X/10
21        Clarity: X/10
22        Accuracy: X/10"""),
23        ("human", "{text}")
24    ])
25    | llm
26    | ScoreParser()
27)
28
29scores = score_chain.invoke({"text": "AI content to evaluate..."})
30print(scores)  # {"relevance": 8, "clarity": 7, "accuracy": 9}

1class Education(BaseModel):
2    degree: str
3    institution: str
4    year: Optional[str] = None
5
6class Experience(BaseModel):
7    company: str
8    role: str
9    duration: str
10    highlights: List[str]
11
12class ResumeData(BaseModel):
13    name: str
14    email: Optional[str] = None
15    skills: List[str]
16    education: List[Education]
17    experience: List[Experience]
18    summary: str
19
20resume_extractor = llm.with_structured_output(ResumeData)
21
22cv_text = """
23Nguyen Van A - Senior Data Engineer
24Email: a.nguyen@email.com
25
26Education:
27- MSc Computer Science, VNU 2020
28- BSc IT, HUST 2018
29
30Experience:
31- Senior Data Engineer at TechCorp (2022-present)
32  Built ETL pipelines, managed data warehouse
33- Data Engineer at StartupXYZ (2020-2022)  
34  Designed data models, implemented Spark jobs
35
36Skills: Python, SQL, Spark, Airflow, AWS, Docker
37"""
38
39resume = resume_extractor.invoke(
40    f"Extract structured data tu CV:\n{cv_text}"
41)

1from langchain_core.exceptions import OutputParserException
2
3def extract_with_retry(chain, text, max_retries=3):
4    for attempt in range(max_retries):
5        try:
6            result = chain.invoke({"text": text})
7            # Validate result
8            if result and hasattr(result, 'name'):
9                return result
10        except OutputParserException as e:
11            print(f"Parse error attempt {attempt + 1}: {e}")
12        except Exception as e:
13            print(f"Error attempt {attempt + 1}: {e}")
14    return None
15
16# With auto-fix
17from langchain.output_parsers import OutputFixingParser
18
19fixing_parser = OutputFixingParser.from_llm(
20    parser=parser,
21    llm=llm
22)
23# Auto-corrects malformed output