from openai import OpenAI
import json

client = OpenAI()

def extract_with_prompt(text: str) -> dict:
    """使用 Prompt 约束提取结构化数据"""
    
    prompt = f"""从以下文本中提取信息，以 JSON 格式返回。

文本：
{text}

请提取以下信息并以 JSON 格式返回：
- name: 人名
- age: 年龄（数字）
- occupation: 职业
- skills: 技能列表

只返回 JSON，不要其他内容。

JSON:"""

    response = client.chat.completions.create(
        model="gpt-4",
        messages=[{"role": "user", "content": prompt}],
        temperature=0  # 降低随机性
    )
    
    # 解析 JSON
    try:
        result = json.loads(response.choices[0].message.content)
        return result
    except json.JSONDecodeError:
        # 尝试提取 JSON 部分
        content = response.choices[0].message.content
        start = content.find('{')
        end = content.rfind('}') + 1
        if start != -1 and end > start:
            return json.loads(content[start:end])
        raise
---

#### 生产级方案：Instructor + Pydantic

在生产环境中，直接解析字符串是非常脆弱的。`Instructor` 是目前最流行的库，它通过 Pydantic 模型定义结构，并自动处理重试逻辑。

#### 核心优势
- **类型安全**：直接返回 Pydantic 对象，支持 IDE 补全。
- **自动重试**：如果 LLM 返回的 JSON 不符合 Schema，Instructor 会自动将错误信息反馈给模型并要求重试。
- **验证逻辑**：支持 Pydantic 的 `Field` 验证和自定义验证器。

#### 代码实现

```python
import instructor
from pydantic import BaseModel, Field, validator
from typing import List, Optional
from openai import OpenAI

# 1. 定义数据模型
class Skill(BaseModel):
    name: str
    level: str = Field(description="技能熟练度，如：入门, 熟练, 精通")

class UserProfile(BaseModel):
    name: str = Field(..., description="用户的真实姓名")
    age: int = Field(..., ge=0, le=120)
    occupation: Optional[str]
    skills: List[Skill]
    
    @validator("name")
    def name_must_be_capitalized(cls, v):
        if not v[0].isupper():
            raise ValueError("姓名首字母必须大写")
        return v

# 2. 封装 Instructor 客户端
client = instructor.patch(OpenAI())

def extract_user_info(text: str) -> UserProfile:
    """使用 Instructor 提取结构化数据"""
    return client.chat.completions.create(
        model="gpt-4o",
        response_model=UserProfile,
        messages=[
            {"role": "system", "content": "你是一个专业的数据提取助手。"},
            {"role": "user", "content": f"提取以下文本中的用户信息：{text}"}
        ],
        max_retries=3  # 自动重试次数
    )

# 示例
text = "张三（Zhang San），今年 28 岁，是一名高级软件工程师。他精通 Python 和 Go 语言。"
try:
    user = extract_user_info(text)
    print(f"姓名: {user.name}, 年龄: {user.age}")
    for skill in user.skills:
        print(f"- 技能: {skill.name} ({skill.level})")
except Exception as e:
    print(f"提取失败: {e}")

from pydantic import BaseModel
from openai import OpenAI

client = OpenAI()

class CalendarEvent(BaseModel):
    name: str
    date: str
    participants: list[str]

# 使用 strict=True 开启严格模式
completion = client.beta.chat.completions.parse(
    model="gpt-4o-2024-08-06",
    messages=[
        {"role": "system", "content": "提取日历事件信息。"},
        {"role": "user", "content": "明天下午 2 点和王总在 3 号会议室开会，讨论 Q4 计划。"},
    ],
    response_format=CalendarEvent,
)

event = completion.choices[0].message.parsed
print(event.name)  # 输出: 会议

import outlines

model = outlines.models.transformers("meta-llama/Meta-Llama-3-8B-Instruct")

# 定义 Schema
prompt = "提取以下文本中的城市和人口：上海有 2400 万人。"
schema = """{
    "title": "CityInfo",
    "type": "object",
    "properties": {
        "city": {"type": "string"},
        "population": {"type": "integer"}
    }
}"""

# 强制生成符合 Schema 的 JSON
generator = outlines.generate.json(model, schema)
result = generator(prompt)
print(result)

from openai import OpenAI

client = OpenAI()

def extract_json_mode(text: str) -> dict:
    """使用 JSON Mode 确保输出为有效 JSON"""
    
    response = client.chat.completions.create(
        model="gpt-4-turbo-preview",
        response_format={"type": "json_object"},  # 启用 JSON Mode
        messages=[
            {
                "role": "system",
                "content": "你是一个信息提取助手，始终以 JSON 格式返回结果。"
            },
            {
                "role": "user",
                "content": f"""从以下文本提取人物信息：

文本：{text}

返回格式：
name
}}"""
            }
        ]
    )
    
    return json.loads(response.choices[0].message.content)

# 使用
text = "赵雷是一名29岁的数据科学家，在阿里巴巴工作3年了。"
result = extract_json_mode(text)
print(result)

def extract_with_schema(text: str) -> dict:
    """使用 JSON Schema 定义结构"""
    
    schema = {
        "type": "object",
        "properties": {
            "name": {"type": "string", "description": "人物姓名"},
            "age": {"type": "integer", "description": "年龄"},
            "occupation": {"type": "string", "description": "职业"},
            "company": {"type": "string", "description": "公司名称"},
            "experience_years": {"type": "integer", "description": "工作年限"}
        },
        "required": ["name", "occupation"]
    }
    
    response = client.chat.completions.create(
        model="gpt-4-turbo-preview",
        response_format={"type": "json_object"},
        messages=[
            {
                "role": "system",
                "content": f"""提取信息并以 JSON 返回。
遵循以下 Schema：
{json.dumps(schema, ensure_ascii=False, indent=2)}"""
            },
            {
                "role": "user",
                "content": f"文本：{text}"
            }
        ]
    )
    
    return json.loads(response.choices[0].message.content)

from openai import OpenAI
from pydantic import BaseModel
from typing import List, Optional

client = OpenAI()

# 定义数据模型
class Person(BaseModel):
    name: str
    age: int
    occupation: str
    skills: List[str]
    company: Optional[str] = None

def extract_structured(text: str) -> Person:
    """使用 Structured Outputs 提取"""
    
    completion = client.beta.chat.completions.parse(
        model="gpt-4o-2024-08-06",
        messages=[
            {
                "role": "system",
                "content": "从文本中提取人物信息。"
            },
            {
                "role": "user",
                "content": text
            }
        ],
        response_format=Person  # 直接传入 Pydantic 模型
    )
    
    return completion.choices[0].message.parsed

# 使用
text = "陈伟是腾讯的高级算法工程师，今年32岁，擅长 NLP 和推荐系统。"
person = extract_structured(text)
print(f"姓名: {person.name}")
print(f"年龄: {person.age}")
print(f"技能: {person.skills}")

from pydantic import BaseModel, Field
from typing import List, Optional
from enum import Enum

class SkillLevel(str, Enum):
    beginner = "beginner"
    intermediate = "intermediate"
    advanced = "advanced"
    expert = "expert"

class Skill(BaseModel):
    name: str
    level: SkillLevel
    years: int = Field(description="使用该技能的年数")

class Education(BaseModel):
    school: str
    degree: str
    major: str
    graduation_year: int

class WorkExperience(BaseModel):
    company: str
    position: str
    start_year: int
    end_year: Optional[int] = None
    responsibilities: List[str]

class Resume(BaseModel):
    """简历结构"""
    name: str
    email: Optional[str] = None
    phone: Optional[str] = None
    summary: str = Field(description="个人简介")
    education: List[Education]
    experience: List[WorkExperience]
    skills: List[Skill]

def parse_resume(resume_text: str) -> Resume:
    """解析简历"""
    
    completion = client.beta.chat.completions.parse(
        model="gpt-4o-2024-08-06",
        messages=[
            {
                "role": "system",
                "content": "你是一个专业的简历解析器，从简历文本中提取结构化信息。"
            },
            {
                "role": "user",
                "content": resume_text
            }
        ],
        response_format=Resume
    )
    
    return completion.choices[0].message.parsed

# 使用
resume_text = """
张三
邮箱：[email protected]
电话：13800138000

个人简介：
5年经验的全栈工程师，专注于 Web 开发和云原生技术。

教育背景：
- 北京大学 计算机科学 硕士 2018年毕业
- 清华大学 软件工程 学士 2016年毕业

工作经历：
1. 字节跳动 高级工程师 2020-至今
   - 负责核心业务系统架构设计
   - 带领5人团队完成微服务改造

2. 阿里巴巴 工程师 2018-2020
   - 开发电商推荐系统
   - 优化系统性能，QPS提升200%

技能：
- Python (专家级，6年)
- JavaScript (高级，5年)
- Kubernetes (中级，3年)
"""

resume = parse_resume(resume_text)
print(f"姓名: {resume.name}")
print(f"教育: {len(resume.education)} 段")
print(f"经历: {len(resume.experience)} 段")

import instructor
from openai import OpenAI
from pydantic import BaseModel, Field
from typing import List

# 初始化 Instructor 客户端
client = instructor.from_openai(OpenAI())

class Product(BaseModel):
    """产品信息"""
    name: str = Field(description="产品名称")
    price: float = Field(description="价格（元）")
    category: str = Field(description="产品类别")
    features: List[str] = Field(description="产品特点")

def extract_product(description: str) -> Product:
    """从描述中提取产品信息"""
    
    product = client.chat.completions.create(
        model="gpt-4",
        response_model=Product,
        messages=[
            {
                "role": "user",
                "content": f"从以下产品描述中提取信息：\n\n{description}"
            }
        ]
    )
    
    return product

# 使用
description = """
新款 iPhone 15 Pro Max 现已上市！售价 9999 元起。
作为苹果最新旗舰手机，它配备了 A17 Pro 芯片、
钛金属边框、4800万像素主摄和 USB-C 接口。
"""

product = extract_product(description)
print(product.model_dump_json(indent=2))

from pydantic import BaseModel, Field, field_validator
from typing import List
import instructor
from openai import OpenAI

client = instructor.from_openai(OpenAI())

class OrderItem(BaseModel):
    product: str
    quantity: int = Field(ge=1, description="数量必须大于0")
    unit_price: float = Field(ge=0, description="单价必须非负")
    
    @property
    def total(self) -> float:
        return self.quantity * self.unit_price

class Order(BaseModel):
    order_id: str = Field(pattern=r"^ORD-\d{6}$", description="订单号格式：ORD-XXXXXX")
    customer_name: str
    items: List[OrderItem]
    
    @field_validator('customer_name')
    @classmethod
    def validate_name(cls, v):
        if len(v) < 2:
            raise ValueError("客户名称至少2个字符")
        return v
    
    @property
    def total_amount(self) -> float:
        return sum(item.total for item in self.items)

def parse_order(text: str) -> Order:
    """解析订单信息"""
    
    return client.chat.completions.create(
        model="gpt-4",
        response_model=Order,
        messages=[
            {
                "role": "system",
                "content": "从文本中提取订单信息。如果没有订单号，生成格式为 ORD-XXXXXX 的订单号。"
            },
            {
                "role": "user",
                "content": text
            }
        ],
        max_retries=3  # 验证失败时自动重试
    )

# 使用
order_text = """
客户张伟下单了以下商品：
- 苹果 x 5个，每个 8 元
- 香蕉 x 3斤，每斤 6 元
- 西瓜 x 1个，15 元
"""

order = parse_order(order_text)
print(f"订单号: {order.order_id}")
print(f"客户: {order.customer_name}")
print(f"总金额: {order.total_amount} 元")

from instructor import Partial
from pydantic import BaseModel
from typing import List

class Article(BaseModel):
    title: str
    summary: str
    key_points: List[str]
    conclusion: str

def stream_article_extraction(text: str):
    """流式提取文章结构"""
    
    for partial_article in client.chat.completions.create_partial(
        model="gpt-4",
        response_model=Article,
        messages=[
            {
                "role": "user",
                "content": f"分析以下文章并提取结构化信息：\n\n{text}"
            }
        ],
        stream=True
    ):
        # partial_article 是部分填充的 Article 对象
        print(f"当前状态: {partial_article}")
        yield partial_article

# 使用
for partial in stream_article_extraction(article_text):
    if partial.title:
        print(f"标题: {partial.title}")
    if partial.key_points:
        print(f"要点数: {len(partial.key_points)}")

from openai import OpenAI
import json

client = OpenAI()

# 定义函数 Schema
tools = [
    {
        "type": "function",
        "function": {
            "name": "extract_meeting_info",
            "description": "从文本中提取会议信息",
            "parameters": {
                "type": "object",
                "properties": {
                    "title": {
                        "type": "string",
                        "description": "会议主题"
                    },
                    "date": {
                        "type": "string",
                        "description": "会议日期，格式 YYYY-MM-DD"
                    },
                    "time": {
                        "type": "string",
                        "description": "会议时间，格式 HH:MM"
                    },
                    "location": {
                        "type": "string",
                        "description": "会议地点"
                    },
                    "participants": {
                        "type": "array",
                        "items": {"type": "string"},
                        "description": "参会人员列表"
                    },
                    "agenda": {
                        "type": "array",
                        "items": {"type": "string"},
                        "description": "会议议程"
                    }
                },
                "required": ["title", "date", "time"]
            }
        }
    }
]

def extract_meeting(text: str) -> dict:
    """使用 Function Calling 提取会议信息"""
    
    response = client.chat.completions.create(
        model="gpt-4",
        messages=[
            {
                "role": "system",
                "content": "你是一个会议信息提取助手。"
            },
            {
                "role": "user",
                "content": f"请从以下文本中提取会议信息：\n\n{text}"
            }
        ],
        tools=tools,
        tool_choice={"type": "function", "function": {"name": "extract_meeting_info"}}
    )
    
    # 解析函数调用结果
    tool_call = response.choices[0].message.tool_calls[0]
    return json.loads(tool_call.function.arguments)

# 使用
text = """
提醒：明天下午2点在3楼会议室有产品评审会。
参会人员：张经理、李工、王设计师
议程：
1. Q3产品规划汇报
2. 新功能演示
3. 讨论下一步计划
"""

meeting = extract_meeting(text)
print(json.dumps(meeting, ensure_ascii=False, indent=2))

tools = [
    {
        "type": "function",
        "function": {
            "name": "extract_person",
            "description": "提取人物信息",
            "parameters": {
                "type": "object",
                "properties": {
                    "name": {"type": "string"},
                    "age": {"type": "integer"},
                    "occupation": {"type": "string"}
                },
                "required": ["name"]
            }
        }
    },
    {
        "type": "function",
        "function": {
            "name": "extract_company",
            "description": "提取公司信息",
            "parameters": {
                "type": "object",
                "properties": {
                    "name": {"type": "string"},
                    "industry": {"type": "string"},
                    "employee_count": {"type": "integer"}
                },
                "required": ["name"]
            }
        }
    },
    {
        "type": "function",
        "function": {
            "name": "extract_event",
            "description": "提取事件信息",
            "parameters": {
                "type": "object",
                "properties": {
                    "event_type": {"type": "string"},
                    "date": {"type": "string"},
                    "description": {"type": "string"}
                },
                "required": ["event_type"]
            }
        }
    }
]

def smart_extract(text: str) -> dict:
    """智能选择合适的提取函数"""
    
    response = client.chat.completions.create(
        model="gpt-4",
        messages=[
            {
                "role": "user",
                "content": f"从以下文本提取关键信息：\n\n{text}"
            }
        ],
        tools=tools,
        tool_choice="auto"  # 让模型自动选择
    )
    
    results = []
    for tool_call in response.choices[0].message.tool_calls:
        results.append({
            "type": tool_call.function.name,
            "data": json.loads(tool_call.function.arguments)
        })
    
    return results

from pydantic import BaseModel
from typing import List, Optional
import instructor
from openai import OpenAI

client = instructor.from_openai(OpenAI())

class Entity(BaseModel):
    """实体"""
    name: str
    type: str  # Person, Organization, Location, etc.
    description: Optional[str] = None

class Relation(BaseModel):
    """关系"""
    subject: str  # 主体实体名称
    predicate: str  # 关系类型
    object: str  # 客体实体名称
    confidence: float = 1.0

class KnowledgeGraph(BaseModel):
    """知识图谱"""
    entities: List[Entity]
    relations: List[Relation]

def extract_knowledge_graph(text: str) -> KnowledgeGraph:
    """从文本构建知识图谱"""
    
    return client.chat.completions.create(
        model="gpt-4",
        response_model=KnowledgeGraph,
        messages=[
            {
                "role": "system",
                "content": """从文本中提取实体和关系，构建知识图谱。
实体类型包括：Person（人物）、Organization（组织）、Location（地点）、Event（事件）、Product（产品）
关系类型包括：works_at（就职于）、located_in（位于）、founded（创立）、acquired（收购）、partners_with（合作）等"""
            },
            {
                "role": "user",
                "content": text
            }
        ]
    )

# 使用
text = """
2023年，阿里巴巴集团在杭州宣布成立达摩院，由张勇担任首席技术官。
达摩院与清华大学建立了战略合作关系，共同研究人工智能技术。
同年，达摩院发布了通义千问大模型，在业界引起广泛关注。
"""

kg = extract_knowledge_graph(text)

print("实体:")
for entity in kg.entities:
    print(f"  - {entity.name} ({entity.type})")

print("\n关系:")
for relation in kg.relations:
    print(f"  - {relation.subject} --[{relation.predicate}]--> {relation.object}")

from pydantic import BaseModel
from typing import List
import instructor
from openai import OpenAI

client = instructor.from_openai(OpenAI())

class TableRow(BaseModel):
    """表格行"""
    values: List[str]

class Table(BaseModel):
    """表格结构"""
    headers: List[str]
    rows: List[TableRow]
    
    def to_dict_list(self) -> List[dict]:
        """转换为字典列表"""
        return [
            dict(zip(self.headers, row.values))
            for row in self.rows
        ]

def extract_table(text: str) -> Table:
    """从文本中提取表格数据"""
    
    return client.chat.completions.create(
        model="gpt-4",
        response_model=Table,
        messages=[
            {
                "role": "user",
                "content": f"从以下文本中提取表格数据：\n\n{text}"
            }
        ]
    )

# 使用
text = """
2023年第三季度各部门业绩：
销售部完成了1200万的销售额，同比增长15%；
技术部完成了8个项目，成本控制在500万以内；
市场部获得了50万新用户，转化率达到12%。
"""

table = extract_table(text)
print("表格数据:")
for row in table.to_dict_list():
    print(row)

# 1. 使用详细的 Field 描述
class Person(BaseModel):
    name: str = Field(
        description="人物的完整姓名，包括姓和名",
        examples=["张三", "李明"]
    )
    age: int = Field(
        ge=0, le=150,
        description="年龄，必须是正整数"
    )

# 2. 提供示例
messages = [
    {
        "role": "system",
        "content": """提取人物信息。

示例输入：小王今年25岁，在北京工作。
示例输出：{"name": "小王", "age": 25, "location": "北京"}"""
    },
    {"role": "user", "content": text}
]

# 3. 分步提取
def extract_complex(text: str):
    # 先提取基本信息
    basic = extract_basic_info(text)
    # 再提取详细信息
    details = extract_details(text, context=basic)
    # 合并结果
    return merge_results(basic, details)

from pydantic import ValidationError
import instructor

def safe_extract(text: str, max_retries: int = 3):
    """带错误处理的提取"""
    
    for attempt in range(max_retries):
        try:
            result = client.chat.completions.create(
                model="gpt-4",
                response_model=MyModel,
                messages=[{"role": "user", "content": text}],
                max_retries=2
            )
            return result
            
        except ValidationError as e:
            print(f"验证错误 (尝试 {attempt + 1}): {e}")
            # 可以修改提示重试
            
        except Exception as e:
            print(f"提取失败: {e}")
            
    return None