Spaces:
Sleeping
Sleeping
| # imports | |
| from typing import TypedDict, Annotated, Optional | |
| from langgraph.graph.message import add_messages | |
| from langchain_core.messages import AnyMessage, SystemMessage | |
| from langchain_openai import AzureChatOpenAI | |
| from langgraph.graph import START, StateGraph | |
| from tools import duckduck_tool, wiki_RAG_tool, image_analyser_tool, audio_transcriber_tool, python_script_opener | |
| from langgraph.prebuilt import ToolNode, tools_condition | |
| import prompts_lib as my_prompts | |
| import os | |
| from dotenv import load_dotenv | |
| # load environment variables | |
| load_dotenv() # take environment variables | |
| # define state | |
| class State(TypedDict): | |
| messages: Annotated[list[AnyMessage], add_messages] | |
| file_path: Optional[str] | |
| # create llm interface | |
| llm = AzureChatOpenAI( | |
| deployment_name = os.environ.get("AZURE_OPENAI_DEPLOYMENT_NAME"), | |
| openai_api_key = os.environ.get("AZURE_OPENAI_API_KEY"), | |
| azure_endpoint = os.environ.get("AZURE_OPENAI_ENDPOINT"), | |
| openai_api_version = os.environ.get("OPENAI_API_VERSION"), | |
| temperature=0 | |
| ) | |
| # bild tools | |
| tools = [duckduck_tool, wiki_RAG_tool, image_analyser_tool, audio_transcriber_tool, python_script_opener] | |
| chat_w_tools = llm.bind_tools(tools) | |
| # load system prompt | |
| system_prompt = my_prompts.system_prompt2 | |
| system_message = SystemMessage(content=system_prompt) | |
| # define nodes | |
| def assistant(state: State): | |
| file_path = state.get("file_path", None) | |
| if file_path: | |
| state["messages"].append(SystemMessage(content=f"File path provided: {file_path}")) | |
| return { | |
| "messages": [chat_w_tools.invoke([system_message] + state["messages"])], | |
| } | |
| # define graph | |
| builder = StateGraph(State) | |
| # add nodes | |
| builder.add_node("assistant", assistant) | |
| builder.add_node("tools", ToolNode(tools)) | |
| # define edges | |
| builder.add_edge(START, "assistant") | |
| builder.add_conditional_edges("assistant", tools_condition, | |
| {"tools": "tools", "__end__": "__end__"}) | |
| builder.add_edge("tools", "assistant") | |
| # compile gtaph | |
| # agent = builder.compile() | |
| class Agent: | |
| def __init__(self): | |
| self.builder = builder.compile() |