from langchain.chat_models import init_chat_model# Follow the steps here to configure your credentials:# https://docs.aws.amazon.com/bedrock/latest/userguide/getting-started.htmlmodel = init_chat_model( "anthropic.claude-3-5-sonnet-20240620-v1:0", model_provider="bedrock_converse",)
conversation = [ {"role": "system", "content": "You are a helpful assistant that translates English to French."}, {"role": "user", "content": "Translate: I love programming."}, {"role": "assistant", "content": "J'adore la programmation."}, {"role": "user", "content": "Translate: I love building applications."}]response = model.invoke(conversation)print(response) # AIMessage("J'adore créer des applications.")
消息对象
from langchain.messages import HumanMessage, AIMessage, SystemMessageconversation = [ SystemMessage("You are a helpful assistant that translates English to French."), HumanMessage("Translate: I love programming."), AIMessage("J'adore la programmation."), HumanMessage("Translate: I love building applications.")]response = model.invoke(conversation)print(response) # AIMessage("J'adore créer des applications.")
full = None # None | AIMessageChunkfor chunk in model.stream("What color is the sky?"): full = chunk if full is None else full + chunk print(full.text)# The# The sky# The sky is# The sky is typically# The sky is typically blue# ...print(full.content_blocks)# [{"type": "text", "text": "The sky is typically blue..."}]
responses = model.batch([ "Why do parrots have colorful feathers?", "How do airplanes fly?", "What is quantum computing?"])for response in responses: print(response)
for response in model.batch_as_completed([ "Why do parrots have colorful feathers?", "How do airplanes fly?", "What is quantum computing?"]): print(response)
from langchain.tools import tool@tooldef get_weather(location: str) -> str: """Get the weather at a location.""" return f"It's sunny in {location}."model_with_tools = model.bind_tools([get_weather])response = model_with_tools.invoke("What's the weather like in Boston?")for tool_call in response.tool_calls: # 查看模型发起的工具调用 print(f"Tool: {tool_call['name']}") print(f"Args: {tool_call['args']}")
# 将（可能多个）工具绑定到模型model_with_tools = model.bind_tools([get_weather])# 步骤 1：模型生成工具调用messages = [{"role": "user", "content": "What's the weather in Boston?"}]ai_msg = model_with_tools.invoke(messages)messages.append(ai_msg)# 步骤 2：执行工具并收集结果for tool_call in ai_msg.tool_calls: # 使用生成的参数执行工具 tool_result = get_weather.invoke(tool_call) messages.append(tool_result)# 步骤 3：将结果传回模型以获取最终响应final_response = model_with_tools.invoke(messages)print(final_response.text)# "The current weather in Boston is 72°F and sunny."
for chunk in model_with_tools.stream( "What's the weather in Boston and Tokyo?"): # 工具调用块逐步到达 for tool_chunk in chunk.tool_call_chunks: if name := tool_chunk.get("name"): print(f"Tool: {name}") if id_ := tool_chunk.get("id"): print(f"ID: {id_}") if args := tool_chunk.get("args"): print(f"Args: {args}")# 输出：# Tool: get_weather# ID: call_SvMlU1TVIZugrFLckFE2ceRE# Args: {"lo# Args: catio# Args: n": "B# Args: osto# Args: n"}# Tool: get_weather# ID: call_QMZdy6qInx13oWKE7KhuhOLR# Args: {"lo# Args: catio# Args: n": "T# Args: okyo# Args: "}
你可以累积块来构建完整的工具调用：
累积工具调用
gathered = Nonefor chunk in model_with_tools.stream("What's the weather in Boston?"): gathered = chunk if gathered is None else gathered + chunk print(gathered.tool_calls)
from pydantic import BaseModel, Fieldclass Movie(BaseModel): """A movie with details.""" title: str = Field(description="The title of the movie") year: int = Field(description="The year the movie was released") director: str = Field(description="The director of the movie") rating: float = Field(description="The movie's rating out of 10")model_with_structure = model.with_structured_output(Movie)response = model_with_structure.invoke("Provide details about the movie Inception")print(response) # Movie(title="Inception", year=2010, director="Christopher Nolan", rating=8.8)
from typing_extensions import TypedDict, Annotatedclass MovieDict(TypedDict): """A movie with details.""" title: Annotated[str, ..., "The title of the movie"] year: Annotated[int, ..., "The year the movie was released"] director: Annotated[str, ..., "The director of the movie"] rating: Annotated[float, ..., "The movie's rating out of 10"]model_with_structure = model.with_structured_output(MovieDict)response = model_with_structure.invoke("Provide details about the movie Inception")print(response) # {'title': 'Inception', 'year': 2010, 'director': 'Christopher Nolan', 'rating': 8.8}
import jsonjson_schema = { "title": "Movie", "description": "A movie with details", "type": "object", "properties": { "title": { "type": "string", "description": "The title of the movie" }, "year": { "type": "integer", "description": "The year the movie was released" }, "director": { "type": "string", "description": "The director of the movie" }, "rating": { "type": "number", "description": "The movie's rating out of 10" } }, "required": ["title", "year", "director", "rating"]}model_with_structure = model.with_structured_output( json_schema, method="json_schema",)response = model_with_structure.invoke("Provide details about the movie Inception")print(response) # {'title': 'Inception', 'year': 2010, ...}
from pydantic import BaseModel, Fieldclass Movie(BaseModel): """A movie with details.""" title: str = Field(description="The title of the movie") year: int = Field(description="The year the movie was released") director: str = Field(description="The director of the movie") rating: float = Field(description="The movie's rating out of 10")model_with_structure = model.with_structured_output(Movie, include_raw=True)response = model_with_structure.invoke("Provide details about the movie Inception")response# {# "raw": AIMessage(...),# "parsed": Movie(title=..., year=..., ...),# "parsing_error": None,# }
示例：嵌套结构
模式可以嵌套：
from pydantic import BaseModel, Fieldclass Actor(BaseModel): name: str role: strclass MovieDetails(BaseModel): title: str year: int cast: list[Actor] genres: list[str] budget: float | None = Field(None, description="Budget in millions USD")model_with_structure = model.with_structured_output(MovieDetails)
response = model.invoke("Create a picture of a cat")print(response.content_blocks)# [# {"type": "text", "text": "Here's a picture of a cat"},# {"type": "image", "base64": "...", "mime_type": "image/jpeg"},# ]
for chunk in model.stream("Why do parrots have colorful feathers?"): reasoning_steps = [r for r in chunk.content_blocks if r["type"] == "reasoning"] print(reasoning_steps if reasoning_steps else chunk.text)
from langchain.chat_models import init_chat_modelmodel = init_chat_model("gpt-5.4-mini")tool = {"type": "web_search"}model_with_tools = model.bind_tools([tool])response = model_with_tools.invoke("What was a positive news story from today?")print(response.content_blocks)
model = init_chat_model( model="gpt-5.4", model_provider="openai").bind(logprobs=True)response = model.invoke("Why do parrots talk?")print(response.response_metadata["logprobs"])
from pydantic import BaseModel, Fieldclass GetWeather(BaseModel): """Get the current weather in a given location""" location: str = Field(description="The city and state, e.g. San Francisco, CA")class GetPopulation(BaseModel): """Get the current population in a given location""" location: str = Field(description="The city and state, e.g. San Francisco, CA")model = init_chat_model(temperature=0)model_with_tools = model.bind_tools([GetWeather, GetPopulation])model_with_tools.invoke( "what's bigger in 2024 LA or NYC", config={"configurable": {"model": "gpt-5.4-mini"}}).tool_calls
