结构化输出总结与多模态案例#

1
# pip install langchain langchain-openai pydantic
2

3
from langchain_openai import ChatOpenAI
4
from pydantic import BaseModel, Field
5

6
# 1. 定义 Pydantic 模型
7
class BookInfo(BaseModel):
8
    """从用户描述中提取书籍信息。"""
9
    title: str = Field(description="书名")
10
    author: str = Field(description="作者")
11
    year: int = Field(description="出版年份")
12
    genre: str = Field(description="书籍类型，如：小说、科幻、历史等")
13

14
# 2. 用 with_structured_output 绑定 Schema
15
llm = ChatOpenAI(model="gpt-4o", temperature=0)
16
structured_llm = llm.with_structured_output(BookInfo)
17

18
# 3. 直接调用，返回值就是 Pydantic 对象
19
result = structured_llm.invoke("刘慈欣的《三体》是2008年出版的科幻小说")
20

21
print(type(result))    # <class '__main__.BookInfo'>
22
print(result.title)    # 三体
23
print(result.author)   # 刘慈欣
24
print(result.year)     # 2008
25
print(result.genre)    # 科幻

1
# pip install langchain langchain-openai
2

3
from langchain_openai import ChatOpenAI
4

5
json_schema = {
6
    "title": "BookInfo",
7
    "description": "从用户描述中提取书籍信息",
8
    "type": "object",
9
    "properties": {
10
        "title": {"type": "string", "description": "书名"},
11
        "author": {"type": "string", "description": "作者"},
12
        "year": {"type": "integer", "description": "出版年份"},
13
        "genre": {"type": "string", "description": "书籍类型"},
14
    },
15
    "required": ["title", "author", "year", "genre"],
16
}
17

18
llm = ChatOpenAI(model="gpt-4o", temperature=0)
19
structured_llm = llm.with_structured_output(json_schema)
20
result = structured_llm.invoke("刘慈欣的《三体》是2008年出版的科幻小说")
21
print(type(result))  # <class 'dict'>
22
print(result)        # {'title': '三体', 'author': '刘慈欣', 'year': 2008, 'genre': '科幻'}

1
# pip install langchain langchain-openai
2

3
from typing import TypedDict, Annotated
4
from langchain_openai import ChatOpenAI
5

6
class BookInfo(TypedDict):
7
    """从用户描述中提取书籍信息。"""
8
    title: Annotated[str, ..., "书名"]
9
    author: Annotated[str, ..., "作者"]
10
    year: Annotated[int, ..., "出版年份"]
11
    genre: Annotated[str, ..., "书籍类型"]
12

13
llm = ChatOpenAI(model="gpt-4o", temperature=0)
14
structured_llm = llm.with_structured_output(BookInfo)
15
result = structured_llm.invoke("刘慈欣的《三体》是2008年出版的科幻小说")
16
print(type(result))  # <class 'dict'>

1
# pip install langchain langchain-openai pydantic
2

3
from langchain_openai import ChatOpenAI
4
from pydantic import BaseModel, Field
5

6
class BookInfo(BaseModel):
7
    title: str = Field(description="书名")
8
    author: str = Field(description="作者")
9

10
llm = ChatOpenAI(model="gpt-4o", temperature=0)
11
structured_llm = llm.with_structured_output(BookInfo, include_raw=True)
12
result = structured_llm.invoke("三体是刘慈欣写的")
13

14
print(result["raw"])            # AIMessage 原始响应对象
15
print(result["parsed"])         # BookInfo(title='三体', author='刘慈欣')
16
print(result["parsing_error"])  # None（解析成功时）

1
# pip install langchain langchain-openai pydantic langchain-core
2

3
from langchain_openai import ChatOpenAI
4
from langchain_core.prompts import ChatPromptTemplate
5
from pydantic import BaseModel, Field
6

7
class Sentiment(BaseModel):
8
    """情感分析结果。"""
9
    text: str = Field(description="原始文本")
10
    sentiment: str = Field(description="情感倾向：positive / negative / neutral")
11
    confidence: float = Field(description="置信度，0-1 之间")
12

13
prompt = ChatPromptTemplate.from_messages([
14
    ("system", "你是一个情感分析专家。"),
15
    ("human", "请分析以下文本的情感：{text}"),
16
])
17
llm = ChatOpenAI(model="gpt-4o", temperature=0)
18

19
# 构建 LCEL 管道：Prompt → 结构化 LLM
20
chain = prompt | llm.with_structured_output(Sentiment)
21
result = chain.invoke({"text": "这家餐厅的菜太好吃了，下次还来！"})
22
# Sentiment(text='...', sentiment='positive', confidence=0.95)

1
from langchain_core.messages import HumanMessage
2

3
msg = HumanMessage(content=[
4
    {"type": "text", "text": "请描述这张图片"},
5
    {"type": "image_url", "image_url": {"url": "https://example.com/cat.jpg"}},
6
])

1
# pip install langchain langchain-openai pydantic
2

3
from pathlib import Path
4
from langchain_openai import ChatOpenAI
5
from pydantic import BaseModel, Field
6

7
class ImageAnalysis(BaseModel):
8
    description: str = Field(description="图片描述")
9
    objects: list[str] = Field(description="识别到的物体")
10

11
def safe_analyze_image(image_path: str, text: str = "请分析这张图片") -> dict:
12
    """带错误处理的图片分析函数。"""
13
    path = Path(image_path)
14
    if not path.exists():
15
        return {"error": f"文件不存在：{image_path}"}
16

17
    size_mb = path.stat().st_size / (1024 * 1024)
18
    if size_mb > 20:
19
        return {"error": f"图片过大：{size_mb:.1f}MB，上限为 20MB"}
20

21
    valid_extensions = {".jpg", ".jpeg", ".png", ".gif", ".webp"}
22
    if path.suffix.lower() not in valid_extensions:
23
        return {"error": f"不支持的格式：{path.suffix}"}
24

25
    try:
26
        llm = ChatOpenAI(model="gpt-4o", temperature=0)
27
        structured_llm = llm.with_structured_output(ImageAnalysis, include_raw=True)
28
        msg = build_multimodal_message(text=text, image_path=image_path)
29
        result = structured_llm.invoke([msg])
30

31
        if result["parsing_error"]:
32
            return {"error": f"解析失败：{result['parsing_error']}"}
33
        return {"data": result["parsed"]}
34
    except Exception as e:
35
        return {"error": f"模型调用失败：{str(e)}"}

1
# pip install langchain langchain-openai langchain-core
2

3
from langchain_core.chat_history import InMemoryChatMessageHistory
4
from langchain_core.runnables.history import RunnableWithMessageHistory
5

6
store = {}
7
def get_session_history(session_id: str):
8
    if session_id not in store:
9
        store[session_id] = InMemoryChatMessageHistory()
10
    return store[session_id]
11

12
chain_with_memory = RunnableWithMessageHistory(chain, get_session_history)

1
# pip install langserve fastapi uvicorn
2

3
from fastapi import FastAPI
4
from langserve import add_routes
5

6
app = FastAPI(title="多模态分析 API")
7
add_routes(app, chain, path="/analyze")
8
# 启动：uvicorn main:app --reload