ChatModels 指南#

1
from langchain_openai import ChatOpenAI
2

3
# 创建模型实例
4
model = ChatOpenAI(
5
    model="gpt-4o-mini",
6
    base_url="http://localhost:8212",
7
    api_key="your-api-key",
8
    temperature=0.7,
9
    max_tokens=1000,
10
    timeout=30
11
)

1
from langchain_core.messages import HumanMessage, SystemMessage, AIMessage
2

3
# 系统消息 - 定义AI的角色和行为
4
system_msg = SystemMessage(content="You are a helpful assistant.")
5

6
# 人类消息 - 用户输入
7
human_msg = HumanMessage(content="Hello, how are you?")
8

9
# AI消息 - AI的回复
10
ai_msg = AIMessage(content="I'm doing well, thank you!")

1
# 简单对话
2
response = model.invoke([HumanMessage(content="Hello!")])
3
print(response.content)
4

5
# 带系统消息的对话
6
messages = [
7
    SystemMessage(content="You are a math tutor."),
8
    HumanMessage(content="What is 2+2?")
9
]
10
response = model.invoke(messages)
11
print(response.content)

1
# 构建对话历史
2
conversation = [
3
    HumanMessage(content="My name is Alice."),
4
    AIMessage(content="Nice to meet you, Alice!"),
5
    HumanMessage(content="What's my name?")
6
]
7

8
response = model.invoke(conversation)
9
print(response.content)  # 应该包含 "Alice"

1
message_batches = [
2
    [HumanMessage(content="Hello 1")],
3
    [HumanMessage(content="Hello 2")],
4
    [HumanMessage(content="Hello 3")]
5
]
6

7
responses = model.batch(message_batches)
8
for i, response in enumerate(responses):
9
    print(f"Response {i+1}: {response.content}")

1
import asyncio
2

3
async def async_chat():
4
    response = await model.ainvoke([HumanMessage(content="Hello async!")])
5
    return response.content
6

7
# 运行异步函数
8
result = asyncio.run(async_chat())
9
print(result)

1
async def async_stream_chat():
2
    async for chunk in model.astream([HumanMessage(content="Count to 5")]):
3
        print(chunk.content, end="", flush=True)
4

5
asyncio.run(async_stream_chat())

1
async def concurrent_requests():
2
    tasks = [
3
        model.ainvoke([HumanMessage(content=f"Request {i}")])
4
        for i in range(5)
5
    ]
6

7
    responses = await asyncio.gather(*tasks)
8
    return [r.content for r in responses]
9

10
results = asyncio.run(concurrent_requests())

1
async def async_batch():
2
    message_batches = [
3
        [HumanMessage(content=f"Batch {i}")]
4
        for i in range(3)
5
    ]
6

7
    responses = await model.abatch(message_batches)
8
    return responses
9

10
results = asyncio.run(async_batch())

1
# 启用流式输出
2
streaming_model = ChatOpenAI(
3
    model="gpt-4o-mini",
4
    streaming=True
5
)
6

7
# 获取流式响应
8
for chunk in streaming_model.stream([HumanMessage(content="Tell me a story")]):
9
    print(chunk.content, end="", flush=True)

1
from langchain_core.callbacks import StreamingStdOutCallbackHandler
2

3
class CustomCallbackHandler(StreamingStdOutCallbackHandler):
4
    def on_llm_new_token(self, token: str, **kwargs) -> None:
5
        print(f"Token: {token}")
6

7
model_with_callback = ChatOpenAI(
8
    model="gpt-4o-mini",
9
    streaming=True,
10
    callbacks=[CustomCallbackHandler()]
11
)

1
# 流式输出 - 实时显示
2
chunks = list(streaming_model.stream([HumanMessage(content="What is AI?")]))
3
streaming_response = "".join([chunk.content for chunk in chunks])
4

5
# 普通输出 - 等待完整响应
6
normal_response = model.invoke([HumanMessage(content="What is AI?")])
7

8
print(f"Streaming: {streaming_response}")
9
print(f"Normal: {normal_response.content}")

1
# 处理 base64 编码的图像
2
import base64
3

4
with open("image.png", "rb") as f:
5
    image_data = base64.b64encode(f.read()).decode("utf-8")
6

7
message = {
8
    "role": "user",
9
    "content": [
10
        {"type": "text", "text": "Describe this image:"},
11
        {
12
            "type": "image",
13
            "source_type": "base64",
14
            "data": image_data,
15
            "mime_type": "image/png"
16
        }
17
    ]
18
}
19

20
response = model.invoke([message])
21
print(response.content)

1
# 处理 URL 图像
2
message = {
3
    "role": "user",
4
    "content": [
5
        {"type": "text", "text": "What's in this image?"},
6
        {
7
            "type": "image",
8
            "source_type": "url",
9
            "url": "https://example.com/image.jpg"
10
        }
11
    ]
12
}
13

14
response = model.invoke([message])

1
# 处理 PDF 文档
2
with open("document.pdf", "rb") as f:
3
    pdf_data = base64.b64encode(f.read()).decode("utf-8")
4

5
message = {
6
    "role": "user",
7
    "content": [
8
        {"type": "text", "text": "Summarize this PDF:"},
9
        {
10
            "type": "file",
11
            "source_type": "base64",
12
            "data": pdf_data,
13
            "mime_type": "application/pdf",
14
            "filename": "document.pdf"
15
        }
16
    ]
17
}
18

19
response = model.invoke([message])

1
# 对比多张图像
2
message = {
3
    "role": "user",
4
    "content": [
5
        {"type": "text", "text": "Compare these images:"},
6
        {"type": "image", "source_type": "url", "url": "image1.jpg"},
7
        {"type": "image", "source_type": "url", "url": "image2.jpg"}
8
    ]
9
}
10

11
response = model.invoke([message])

1
from langchain_core.tools import tool
2

3
@tool
4
def add_numbers(a: int, b: int) -> int:
5
    """Add two numbers together.
6

7
    Args:
8
        a: First number
9
        b: Second number
10

11
    Returns:
12
        Sum of the two numbers
13
    """
14
    return a + b
15

16
@tool
17
def get_weather(location: str) -> dict:
18
    """Get weather information for a location.
19

20
    Args:
21
        location: City name
22

23
    Returns:
24
        Weather data dictionary
25
    """
26
    return {
27
        "location": location,
28
        "temperature": 22,
29
        "description": "sunny"
30
    }

1
# 绑定工具到模型
2
model_with_tools = model.bind_tools([add_numbers, get_weather])
3

4
# 请求计算
5
response = model_with_tools.invoke([
6
    HumanMessage(content="What is 15 + 27?")
7
])
8

9
# 检查工具调用
10
if response.tool_calls:
11
    for tool_call in response.tool_calls:
12
        print(f"Tool: {tool_call['name']}")
13
        print(f"Args: {tool_call['args']}")

1
from langchain_core.messages import ToolMessage
2

3
def complete_tool_calling(query: str):
4
    # 1. 发送查询
5
    messages = [HumanMessage(content=query)]
6

7
    # 2. 模型决定调用工具
8
    ai_response = model_with_tools.invoke(messages)
9
    messages.append(ai_response)
10

11
    # 3. 执行工具调用
12
    for tool_call in ai_response.tool_calls:
13
        if tool_call["name"] == "add_numbers":
14
            result = add_numbers.invoke(tool_call)
15
        elif tool_call["name"] == "get_weather":
16
            result = get_weather.invoke(tool_call)
17

18
        # 4. 添加工具结果
19
        tool_message = ToolMessage(
20
            content=str(result.content),
21
            tool_call_id=tool_call["id"],
22
            name=tool_call["name"]
23
        )
24
        messages.append(tool_message)
25

26
    # 5. 获取最终回答
27
    final_response = model_with_tools.invoke(messages)
28
    return final_response.content
29

30
# 使用示例
31
result = complete_tool_calling("Add 15 and 27, then tell me the weather in Beijing")

1
response = model_with_tools.invoke([
2
    HumanMessage(content="Calculate 5+3 and get weather for Tokyo")
3
])
4

5
# 可能同时调用 add_numbers 和 get_weather
6
for tool_call in response.tool_calls:
7
    print(f"Parallel tool call: {tool_call['name']}")

1
@tool
2
def search_database(query: str, table: str) -> list:
3
    """Search database for information.
4

5
    Args:
6
        query: Search query
7
        table: Database table name
8

9
    Returns:
10
        List of matching records
11
    """
12
    # 模拟数据库搜索
13
    return [{"id": 1, "name": "Result 1"}]
14

15
@tool
16
def send_email(recipient: str, subject: str, body: str) -> bool:
17
    """Send an email.
18

19
    Args:
20
        recipient: Email recipient
21
        subject: Email subject
22
        body: Email body
23

24
    Returns:
25
        Success status
26
    """
27
    print(f"Email sent to {recipient}")
28
    return True

1
from pydantic import BaseModel, Field
2

3
class WeatherInfo(BaseModel):
4
    """天气信息结构"""
5
    location: str = Field(description="地点名称")
6
    temperature: float = Field(description="温度")
7
    humidity: int = Field(description="湿度百分比")
8
    description: str = Field(description="天气描述")
9

10
# 创建结构化输出模型
11
structured_model = model.with_structured_output(WeatherInfo)
12

13
# 获取结构化响应
14
weather = structured_model.invoke([
15
    HumanMessage(content="Tell me about the weather in Beijing")
16
])
17

18
print(f"Location: {weather.location}")
19
print(f"Temperature: {weather.temperature}°C")

1
class Person(BaseModel):
2
    name: str = Field(description="人员姓名")
3
    age: int = Field(description="人员年龄")
4
    skills: list[str] = Field(description="技能列表")
5

6
class Team(BaseModel):
7
    team_name: str = Field(description="团队名称")
8
    members: list[Person] = Field(description="团队成员")
9
    project: str = Field(description="项目名称")
10

11
team_model = model.with_structured_output(Team)
12
result = team_model.invoke([
13
    HumanMessage(content="Create a 3-person development team for a web project")
14
])
15

16
print(f"Team: {result.team_name}")
17
for member in result.members:
18
    print(f"- {member.name}, {member.age}, Skills: {member.skills}")

1
from langchain_core.output_parsers import PydanticToolsParser
2

3
@tool
4
def calculate_area(length: float, width: float) -> dict:
5
    """Calculate rectangle area."""
6
    return {"area": length * width}
7

8
tools = [calculate_area]
9
parser = PydanticToolsParser(tools=tools)
10

11
model_with_parser = model.bind_tools(tools) | parser

1
def manage_long_conversation(messages: list, max_tokens: int = 4000):
2
    """管理长对话上下文"""
3
    # 估算token数量（简化计算）
4
    total_tokens = sum(len(msg.content.split()) for msg in messages)
5

6
    if total_tokens > max_tokens:
7
        # 保留系统消息和最近的对话
8
        system_msgs = [msg for msg in messages if isinstance(msg, SystemMessage)]
9
        recent_msgs = messages[-10:]  # 保留最近10条消息
10
        messages = system_msgs + recent_msgs
11

12
    return model.invoke(messages)

1
import requests
2
from openai import APIConnectionError
3

4
def robust_invoke(messages, max_retries=3):
5
    """带重试机制的模型调用"""
6
    for attempt in range(max_retries):
7
        try:
8
            return model.invoke(messages)
9
        except APIConnectionError as e:
10
            if attempt == max_retries - 1:
11
                raise e
12
            print(f"Connection failed, retrying... ({attempt + 1}/{max_retries})")
13
            time.sleep(2 ** attempt)  # 指数退避

1
from openai import AuthenticationError, RateLimitError, BadRequestError
2

3
def handle_api_errors(messages):
4
    """处理各种API错误"""
5
    try:
6
        return model.invoke(messages)
7
    except AuthenticationError:
8
        print("API密钥无效，请检查配置")
9
    except RateLimitError:
10
        print("达到速率限制，请稍后重试")
11
    except BadRequestError as e:
12
        print(f"请求参数错误: {e}")
13
    except Exception as e:
14
        print(f"未知错误: {e}")

1
# 配置超时参数
2
timeout_model = ChatOpenAI(
3
    model="gpt-4o-mini",
4
    timeout=30,  # 30秒超时
5
    max_retries=3
6
)
7

8
try:
9
    response = timeout_model.invoke(messages)
10
except TimeoutError:
11
    print("请求超时，请检查网络连接")

1
def check_context_length(messages, max_tokens=8000):
2
    """检查上下文长度是否超限"""
3
    # 简化的token计算
4
    total_tokens = sum(len(msg.content.split()) * 1.3 for msg in messages)
5

6
    if total_tokens > max_tokens:
7
        raise ValueError(f"Context too long: {total_tokens} > {max_tokens}")
8

9
    return model.invoke(messages)

1
response = model.invoke([HumanMessage(content="Hello")])
2

3
# 访问元数据
4
metadata = response.response_metadata
5
print(f"Model: {metadata.get('model_name')}")
6
print(f"Finish reason: {metadata.get('finish_reason')}")
7

8
# Token使用信息
9
if 'token_usage' in metadata:
10
    usage = metadata['token_usage']
11
    print(f"Prompt tokens: {usage.get('prompt_tokens')}")
12
    print(f"Completion tokens: {usage.get('completion_tokens')}")
13
    print(f"Total tokens: {usage.get('total_tokens')}")

1
def calculate_cost(response):
2
    """计算API调用成本"""
3
    metadata = response.response_metadata
4
    token_usage = metadata.get('token_usage', {})
5

6
    # OpenAI pricing (示例价格)
7
    prompt_cost = token_usage.get('prompt_tokens', 0) * 0.00015 / 1000
8
    completion_cost = token_usage.get('completion_tokens', 0) * 0.0006 / 1000
9

10
    return {
11
        'prompt_cost': prompt_cost,
12
        'completion_cost': completion_cost,
13
        'total_cost': prompt_cost + completion_cost,
14
        'total_tokens': token_usage.get('total_tokens', 0)
15
    }
16

17
response = model.invoke([HumanMessage(content="Explain quantum computing")])
18
cost_info = calculate_cost(response)
19
print(f"Total cost: ${cost_info['total_cost']:.6f}")

1
import time
2

3
def monitor_performance(messages):
4
    """监控模型性能"""
5
    start_time = time.time()
6
    response = model.invoke(messages)
7
    end_time = time.time()
8

9
    response_time = end_time - start_time
10
    metadata = response.response_metadata
11
    token_usage = metadata.get('token_usage', {})
12

13
    return {
14
        'response_time': response_time,
15
        'tokens_per_second': token_usage.get('completion_tokens', 0) / response_time,
16
        'finish_reason': metadata.get('finish_reason'),
17
        'model': metadata.get('model_name')
18
    }
19

20
perf_data = monitor_performance([HumanMessage(content="Write a short story")])
21
print(f"Response time: {perf_data['response_time']:.2f}s")
22
print(f"Tokens/sec: {perf_data['tokens_per_second']:.1f}")

1
from langchain_core.callbacks import get_usage_metadata_callback
2

3
# 使用 context manager 自动追踪
4
with get_usage_metadata_callback() as cb:
5
    # 执行多个 LLM 调用
6
    response1 = model.invoke([HumanMessage(content="Hello")])
7
    response2 = model.invoke([HumanMessage(content="How are you?")])
8
    response3 = model.invoke([HumanMessage(content="Goodbye")])
9

10
    # 获取聚合的 token 使用情况
11
    total_usage = cb.usage_metadata
12

13
# 分析 token 使用
14
for model_name, usage_data in total_usage.items():
15
    print(f"模型: {model_name}")
16
    print(f"  输入tokens: {usage_data['input_tokens']}")
17
    print(f"  输出tokens: {usage_data['output_tokens']}")
18
    print(f"  总计: {usage_data['total_tokens']}")
19

20
    # 详细信息（如果可用）
21
    if 'input_token_details' in usage_data:
22
        print(f"  输入详情: {usage_data['input_token_details']}")
23
    if 'output_token_details' in usage_data:
24
        print(f"  输出详情: {usage_data['output_token_details']}")

1
# 执行管道，所有 token 使用都会被自动追踪
2
results = pipeline.invoke({"topic": "人工智能在教育中的应用"})
3

4
# 获取总的 token 使用情况
5
total_usage = cb.usage_metadata
6

7
# 显示详细统计
8
print("\n📊 Token使用统计:")
9
total_input = 0
10
total_output = 0
11
total_all = 0
12

13
for model_name, usage_data in total_usage.items():
14
    input_tokens = usage_data.get('input_tokens', 0)
15
    output_tokens = usage_data.get('output_tokens', 0)
16
    total_tokens = usage_data.get('total_tokens', 0)
17

18
    print(f"\n模型: {model_name}")
19
    print(f"  输入tokens: {input_tokens}")
20
    print(f"  输出tokens: {output_tokens}")
21
    print(f"  总tokens: {total_tokens}")
22

23
    total_input += input_tokens
24
    total_output += output_tokens
25
    total_all += total_tokens
26

27
print(f"\n🎯 整个管道汇总:")
28
print(f"  总输入tokens: {total_input}")
29
print(f"  总输出tokens: {total_output}")
30
print(f"  总计tokens: {total_all}")

1
# 分别执行每个步骤并实时追踪
2
step_results = {}
3
step_tokens = {}
4
test_topic = "人工智能在教育中的应用"
5

6
# 步骤1：生成大纲
7
print("\n🚀 步骤1: 生成主题大纲...")
8
with get_usage_metadata_callback() as cb1:
9
    outline = (topic_expander | model | StrOutputParser()).invoke({"topic": test_topic})
10
    step_results["step1_outline"] = outline
11
    step_tokens["step1"] = dict(cb1.usage_metadata)
12

13
    print(f"✅ 大纲生成完成 ({len(outline.split())} 词)")
14
    if cb1.usage_metadata:
15
        for model_name, usage in cb1.usage_metadata.items():
16
            print(f"   Token使用 - 输入: {usage.get('input_tokens', 0)}, "
17
                  f"输出: {usage.get('output_tokens', 0)}, "
18
                  f"总计: {usage.get('total_tokens', 0)}")
19

20
# 步骤2：生成内容
21
print("\n🚀 步骤2: 基于大纲生成文章内容...")
22
with get_usage_metadata_callback() as cb2:
23
    content = (content_generator | model | StrOutputParser()).invoke({"outline": outline})
24
    step_results["step2_content"] = content
25
    step_tokens["step2"] = dict(cb2.usage_metadata)
26

27
    print(f"✅ 文章内容生成完成 ({len(content.split())} 词)")
28
    if cb2.usage_metadata:
29
        for model_name, usage in cb2.usage_metadata.items():
30
            print(f"   Token使用 - 输入: {usage.get('input_tokens', 0)}, "
31
                  f"输出: {usage.get('output_tokens', 0)}, "
32
                  f"总计: {usage.get('total_tokens', 0)}")
33

34
# 步骤3：优化内容
35
print("\n🚀 步骤3: 优化文章内容...")
36
with get_usage_metadata_callback() as cb3:
37
    optimized_content = (content_optimizer | model | StrOutputParser()).invoke({"content": content})
38
    step_results["step3_optimized"] = optimized_content
39
    step_tokens["step3"] = dict(cb3.usage_metadata)
40

41
    print(f"✅ 内容优化完成 ({len(optimized_content.split())} 词)")
42
    if cb3.usage_metadata:
43
        for model_name, usage in cb3.usage_metadata.items():
44
            print(f"   Token使用 - 输入: {usage.get('input_tokens', 0)}, "
45
                  f"输出: {usage.get('output_tokens', 0)}, "
46
                  f"总计: {usage.get('total_tokens', 0)}")

1
from langchain_core.callbacks import UsageMetadataCallbackHandler
2

3
def track_step_tokens(step_name: str, step_callback: UsageMetadataCallbackHandler):
4
    """自定义token追踪函数
5

6
    Args:
7
        step_name: 步骤名称，用于标识不同的处理步骤
8
        step_callback: 用于收集token使用数据的回调处理器
9

10
    Returns:
11
        包装后的处理函数，会自动追踪token使用
12
    """
13
    def wrapper(x):
14
        # 根据步骤名称执行对应的链
15
        if step_name == "outline":
16
            chain = topic_expander | model | StrOutputParser()
17
        elif step_name == "content":
18
            chain = content_generator | model | StrOutputParser()
19
        elif step_name == "optimized":
20
            chain = content_optimizer | model | StrOutputParser()
21

22
        # 使用独立的callback追踪这一步
23
        result = chain.invoke(x, config={"callbacks": [step_callback]})
24

25
        # 打印这一步的token使用情况
26
        print(f"\n🔍 步骤 [{step_name}] Token使用情况:")
27
        if step_callback.usage_metadata:
28
            for model_name, usage in step_callback.usage_metadata.items():
29
                print(f"  模型: {model_name}")
30
                print(f"  输入tokens: {usage.get('input_tokens', 0)}")
31
                print(f"  输出tokens: {usage.get('output_tokens', 0)}")
32
                print(f"  总tokens: {usage.get('total_tokens', 0)}")
33

34
        return result
35

36
    return wrapper
37

38
# 使用自定义追踪函数
39
from langchain_schema.runnable import RunnableLambda
40

41
step1_callback = UsageMetadataCallbackHandler()
42
step2_callback = UsageMetadataCallbackHandler()
43
step3_callback = UsageMetadataCallbackHandler()
44

45
# 构建包含token追踪的管道
46
detailed_pipeline = (
47
    RunnablePassthrough.assign(
48
        step1_outline=RunnableLambda(track_step_tokens("outline", step1_callback))
49
    )
50
    | RunnablePassthrough.assign(
51
        step2_content=RunnableLambda(track_step_tokens("content", step2_callback))
52
    )
53
    | RunnablePassthrough.assign(
54
        step3_optimized=RunnableLambda(track_step_tokens("optimized", step3_callback))
55
    )
56
    | RunnablePassthrough.assign(
57
        # 将token使用信息添加到结果中
58
        token_usage=RunnableLambda(lambda x: {
59
            "step1_outline": dict(step1_callback.usage_metadata),
60
            "step2_content": dict(step2_callback.usage_metadata),
61
            "step3_optimized": dict(step3_callback.usage_metadata)
62
        })
63
    )
64
)

1
def analyze_token_efficiency(total_usage: dict) -> dict:
2
    """分析token使用效率
3

4
    Args:
5
        total_usage: get_usage_metadata_callback返回的使用数据
6

7
    Returns:
8
        包含效率指标的分析结果
9
    """
10
    analysis = {}
11

12
    for model_name, usage_data in total_usage.items():
13
        input_tokens = usage_data.get('input_tokens', 0)
14
        output_tokens = usage_data.get('output_tokens', 0)
15
        total_tokens = usage_data.get('total_tokens', 0)
16

17
        # 计算效率指标
18
        if input_tokens > 0:
19
            efficiency_ratio = output_tokens / input_tokens
20
            cost_per_output = input_tokens * 0.15 + output_tokens * 0.60  # gpt-4o-mini价格
21

22
            analysis[model_name] = {
23
                'efficiency_ratio': efficiency_ratio,  # 输出/输入比率
24
                'cost_estimate_usd': cost_per_output / 1_000_000,  # 预估成本
25
                'tokens_per_dollar': total_tokens / (cost_per_output / 1_000_000) if cost_per_output > 0 else 0
26
            }
27

28
    return analysis
29

30
# 使用示例
31
with get_usage_metadata_callback() as cb:
32
    response = model.invoke([HumanMessage(content="写一篇关于AI的短文")])
33
    usage = cb.usage_metadata
34

35
efficiency = analyze_token_efficiency(usage)
36
for model_name, metrics in efficiency.items():
37
    print(f"\n模型: {model_name}")
38
    print(f"  效率比率: {metrics['efficiency_ratio']:.2f}")
39
    print(f"  预估成本: ${metrics['cost_estimate_usd']:.6f}")
40
    print(f"  性价比: {metrics['tokens_per_dollar']:.0f} tokens/美元")

1
from langchain_core.messages import convert_to_openai_messages
2

3
# LangChain 消息
4
messages = [
5
    SystemMessage("You are a helpful assistant."),
6
    HumanMessage("Hello!"),
7
    AIMessage("Hi there!")
8
]
9

10
# 转换为 OpenAI 格式
11
openai_messages = convert_to_openai_messages(messages)
12
print(openai_messages)
13
# 输出:
14
# [
15
#   {'role': 'system', 'content': 'You are a helpful assistant.'},
16
#   {'role': 'user', 'content': 'Hello!'},
17
#   {'role': 'assistant', 'content': 'Hi there!'}
18
# ]

1
# 包含工具调用的消息
2
tool_messages = [
3
    HumanMessage("Calculate 15 + 27"),
4
    AIMessage("", tool_calls=[{
5
        "name": "add_numbers",
6
        "args": {"a": 15, "b": 27},
7
        "id": "call_123"
8
    }]),
9
    ToolMessage("42", tool_call_id="call_123", name="add_numbers")
10
]
11

12
openai_format = convert_to_openai_messages(tool_messages)

1
# 生产环境推荐配置
2
production_model = ChatOpenAI(
3
    model="gpt-4o-mini",
4
    temperature=0.7,           # 平衡创造性和确定性
5
    max_tokens=1000,           # 控制响应长度
6
    timeout=60,                # 合理的超时时间
7
    max_retries=3,             # 自动重试
8
    streaming=True             # 提升用户体验
9
)

1
def process_multimodal_safely(text, image_data=None):
2
    """安全处理多模态输入"""
3
    content = [{"type": "text", "text": text}]
4

5
    if image_data:
6
        # 检查图像大小
7
        if len(image_data) > 20 * 1024 * 1024:  # 20MB limit
8
            raise ValueError("Image too large")
9

10
        content.append({
11
            "type": "image",
12
            "source_type": "base64",
13
            "data": image_data,
14
            "mime_type": "image/jpeg"
15
        })
16

17
    return model.invoke([{"role": "user", "content": content}])