Код: Выделить всё
import getpass
import os
from dotenv import load_dotenv
from typing import Annotated, List, TypedDict
import functools
import operator
from pathlib import Path
from tempfile import TemporaryDirectory
from langchain_core.tools import tool
from langchain.agents import AgentExecutor, create_openai_functions_agent
from langchain.output_parsers.openai_functions import JsonOutputFunctionsParser
from langchain_core.prompts import ChatPromptTemplate, MessagesPlaceholder
from langchain_core.messages import BaseMessage, HumanMessage
from langchain_aws import ChatBedrock
from langchain_core.pydantic_v1 import BaseModel, Field
from langgraph.graph import END, StateGraph, START
load_dotenv()
llm = ChatBedrock(
model_id="anthropic.claude-3-sonnet-20240229-v1:0",
model_kwargs=dict(temperature=0),
region_name="us-east-1",
credentials_profile_name="test"
)
@tool
def hotel_search_by_destination():
"""
Search for hotels in the given destination.
"""
pass
@tool
def hotel_search_by_name():
"""
Search for hotels by name.
"""
pass
@tool
def hotel_search_by_id():
"""
Search for hotels by id.
"""
pass
@tool
def search_flights():
"""
Search for flights.
"""
pass
@tool
def search_tickets():
"""
Search for tickets.
"""
pass
def create_agent(llm: str, tools: list, system_prompt: str):
"""
Create a function calling agent and add it to the graph.
"""
system_prompt += "\nWork autonomously according to your specialty, using the tools available to you."
" Do not ask for clarification."
" Your other team members (and other teams) will collaborate with you with their own specialties."
" You are chosen for a reason! You are one of the following team members: {team_members}."
prompt = ChatPromptTemplate.from_messages(
[
("system", system_prompt),
MessagesPlaceholder(variable_name="messages"),
MessagesPlaceholder(variable_name="agent_scratchpad"),
]
)
agent = create_openai_functions_agent(llm, tools, prompt)
executor = AgentExecutor(agent=agent, tools=tools)
return executor
def agent_node(state, agent, name):
result = agent.invoke(state)
return {"messages": [HumanMessage(content=result["output"], name=name)]}
def create_team_supervisor(llm: ChatBedrock, system_prompt, members) -> str:
"""An LLM-based router"""
options = ["FINISH"] + members
function_def = {
"name": "route",
"description": "Select the next role.",
"parameters": {
"title": "routeSchema",
"type": "object",
"properties": {
"next": {
"title": "Next",
"anyOf": [
{"enum": options},
],
},
},
"required": ["next"],
},
}
prompt = ChatPromptTemplate.from_messages(
[
("system", system_prompt),
MessagesPlaceholder(variable_name="messages"),
(
"system",
"Given the conversation above, who should act next?"
" Or should we FINISH? Select one of: {options}",
),
]
).partial(options=str(options), team_members=", ".join(members))
return (
prompt
| llm.bind_tools(functions=[function_def], function_call="route", tools=[function_def])
| JsonOutputFunctionsParser()
)
class HotelTeamState(TypedDict):
messages: Annotated[List[BaseMessage], operator.add]
team_members: List[str]
next: str
llm = ChatBedrock(
model_id="anthropic.claude-3-sonnet-20240229-v1:0",
model_kwargs=dict(temperature=0),
)
hotel_search_by_destination_agent = create_agent(
llm,
[hotel_search_by_destination],
"You are a research assistant who can search hotels by destination.",
)
hotel_search_by_destination_node = functools.partial(agent_node, agent=hotel_search_by_destination_agent, name="SearchHotelsByDestination")
hotel_search_by_name_agent = create_agent(
llm,
[hotel_search_by_name],
"You are a research assistant who can search hotels by destination.",
)
hotel_search_by_name_node = functools.partial(agent_node, agent=hotel_search_by_name_agent, name="SearchHotelsByName")
hotel_search_by_id_agent = create_agent(
llm,
[hotel_search_by_id],
"You are a research assistant who can search hotels by destination.",
)
hotel_search_by_id_node = functools.partial(agent_node, agent=hotel_search_by_id_agent, name="SearchHotelsById")
supervisor_agent = create_team_supervisor(
llm,
"You are a supervisor tasked with managing a conversation between the"
" following workers: SearchHotelsByDestination, SearchHotelsByName and SearchHotelsById. Given the following user request,"
" respond with the worker to act next. Each worker will perform a"
" task and respond with their results and status. When finished,"
" respond with FINISH.",
["SearchHotelsByDestination", "SearchHotelsByName", "SearchHotelsById"],
)
research_graph = StateGraph(HotelTeamState)
research_graph.add_node("SearchHotelsByDestination", hotel_search_by_destination_node)
research_graph.add_node("SearchHotelsByName", hotel_search_by_name_node)
research_graph.add_node("SearchHotelsById", hotel_search_by_id_node)
research_graph.add_node("supervisor", supervisor_agent)
# Define the control flow
research_graph.add_edge("SearchHotelsByDestination", "supervisor")
research_graph.add_edge("SearchHotelsByName", "supervisor")
research_graph.add_edge("SearchHotelsById", "supervisor")
research_graph.add_conditional_edges(
"supervisor",
lambda x: x["next"],
{
"SearchHotelsByDestination": "SearchHotelsByDestination",
"SearchHotelsByName": "SearchHotelsByName",
"SearchHotelsById": "SearchHotelsById",
"FINISH": END,
},
)
research_graph.add_edge(START, "supervisor")
chain = research_graph.compile()
# The following functions interoperate between the top level graph state
# and the state of the research sub-graph
# this makes it so that the states of each graph don't get intermixed
def enter_chain(message: str):
results = {
"messages": [
HumanMessage(content=message),
],
}
return results
hotels_chain = enter_chain | chain
for s in hotels_chain.stream(
"what are the available hotels in madrid", {"recursion_limit": 100}
):
if "__end__" not in s:
print(s)
print("---")
Проблема связана с тем, что я использую Amazon Bedrock!?
Я пытался видели разные подходы, но ни один из них не сработал
Подробнее здесь: https://stackoverflow.com/questions/787 ... zonbedrock
