Issue with executing Langchain's LLM Compiler example code #27439
Replies: 1 comment 10 replies
-
|
I found a similar unsolved discussion that might be related to your issue: When i run my demo,It just stuck at llm.invoke().I can't figure what wrong with it [1]. Additionally, there was an issue titled ainvoke take along time? which was closed without comments or a solution [2]. Regarding your specific problem, there are a few potential reasons why
You might want to add logging or print statements to trace the execution flow and identify where exactly the process is getting stuck. This can help in diagnosing the issue more effectively. |
Beta Was this translation helpful? Give feedback.
-
|
To implement a Plan and Execute agent using the LLM Compiler version in LangChain, you can follow these steps:
Here's a sample implementation: from langchain_core.language_models import BaseChatModel
from langchain_core.messages import HumanMessage, SystemMessage, AIMessage
from langchain_core.prompts import ChatPromptTemplate
from langchain_core.runnables import RunnableBranch
from langchain_core.tools import BaseTool
from langchain_core.output_parsers import StrOutputParser
from langchain_community.tools import SQLDatabase
from langchain_openai import ChatOpenAI
from typing import List, Dict, Any, Union
from concurrent.futures import ThreadPoolExecutor, wait
# Define the LLM
llm = ChatOpenAI(model="gpt-4-turbo-preview")
# Define the SQL Database connection
db = SQLDatabase.from_uri("postgresql://username:password@localhost/dbname")
# Define the tools
class SQLTool(BaseTool):
def invoke(self, query: str):
# Execute the SQL query against the database
return db.execute(query)
tools = [SQLTool(name="SQLExecutor", description="Executes SQL queries")]
# Planner: Create a plan of tasks
def planner(question: str) -> List[Dict[str, Any]]:
# Example plan: Generate SQL, Validate SQL, Execute SQL
return [
{"idx": 1, "tool": "GenerateSQL", "args": {"question": question}},
{"idx": 2, "tool": "ValidateSQL", "args": {"query": "$1"}},
{"idx": 3, "tool": "SQLExecutor", "args": {"query": "$2"}}
]
# Task Fetching Unit: Execute tasks
def execute_task(task, observations):
tool_name = task["tool"]
args = task["args"]
resolved_args = {k: observations.get(v.strip("$"), v) for k, v in args.items()}
tool = next(tool for tool in tools if tool.name == tool_name)
return tool.invoke(resolved_args["query"])
def task_fetching_unit(tasks: List[Dict[str, Any]]) -> Dict[int, Any]:
observations = {}
with ThreadPoolExecutor() as executor:
futures = {executor.submit(execute_task, task, observations): task for task in tasks}
for future in wait(futures).done:
task = futures[future]
observations[task["idx"]] = future.result()
return observations
# Joiner: Respond to the user
def joiner(observations: Dict[int, Any]) -> str:
# Combine results and decide on the final response
final_result = observations.get(3, "No result")
return f"Final result: {final_result}"
# Plan and Execute function
def plan_and_execute(question: str) -> str:
tasks = planner(question)
observations = task_fetching_unit(tasks)
return joiner(observations)
# Example usage
question = "What is the total sales for last month?"
result = plan_and_execute(question)
print(result)Key Components:
This code provides a basic structure for implementing a Plan and Execute agent using the LLM Compiler version. You can expand it by adding more sophisticated task planning, error handling, and result processing as needed. |
Beta Was this translation helpful? Give feedback.
-
|
@dosu but it doesn't have LangGraph/LangFuse. Something where I create a StateGraph workflow like: The LLM Compiler based plan and execute should have a LangGraph Stategraph and nodes and edges for parse_question, generate_sql, validate_sql, execute_sql |
Beta Was this translation helpful? Give feedback.
-
|
To create a Plan and Execute agent using LangGraph with a from langgraph.graph import StateGraph, START, END
from langgraph.checkpoint.memory import MemorySaver
from langchain_core.language_models import BaseChatModel
from langchain_openai import ChatOpenAI
from langchain_community.tools import SQLDatabase
from typing import Any, Dict
# Define the LLM
llm = ChatOpenAI(model="gpt-4-turbo-preview")
# Define the SQL Database connection
db = SQLDatabase.from_uri("postgresql://username:password@localhost/dbname")
# Define the SQL agent with necessary methods
class SQLAgent:
def parse_question(self, state: Dict[str, Any], memory: MemorySaver):
# Logic to parse the question
question = state["input"]
# Store parsed question in memory
memory.save("parsed_question", question)
return {"parsed_question": question}
def generate_sql(self, state: Dict[str, Any], memory: MemorySaver):
# Logic to generate SQL from parsed question
parsed_question = memory.load("parsed_question")
sql_query = f"SELECT * FROM sales WHERE month = '{parsed_question}'"
memory.save("sql_query", sql_query)
return {"sql_query": sql_query}
def validate_sql(self, state: Dict[str, Any], memory: MemorySaver):
# Logic to validate the generated SQL
sql_query = memory.load("sql_query")
# Assume validation is successful
return {"validated_sql": sql_query}
def execute_sql(self, state: Dict[str, Any], memory: MemorySaver):
# Logic to execute the SQL query
validated_sql = memory.load("validated_sql")
result = db.execute(validated_sql)
return {"result": result}
# Workflow manager to create the StateGraph
class WorkflowManager:
def __init__(self):
self.sql_agent = SQLAgent()
self.memory = MemorySaver()
def create_workflow(self) -> StateGraph:
workflow = StateGraph()
# Add nodes to the graph
workflow.add_node("parse_question", lambda state: self.sql_agent.parse_question(state, self.memory))
workflow.add_node("generate_sql", lambda state: self.sql_agent.generate_sql(state, self.memory))
workflow.add_node("validate_sql", lambda state: self.sql_agent.validate_sql(state, self.memory))
workflow.add_node("execute_sql", lambda state: self.sql_agent.execute_sql(state, self.memory))
# Define edges
workflow.add_edge(START, "parse_question")
workflow.add_edge("parse_question", "generate_sql")
workflow.add_edge("generate_sql", "validate_sql")
workflow.add_edge("validate_sql", "execute_sql")
workflow.add_edge("execute_sql", END)
return workflow
# Example usage
workflow_manager = WorkflowManager()
workflow = workflow_manager.create_workflow()
app = workflow.compile(checkpointer=workflow_manager.memory)
# Input state
input_state = {"input": "last month"}
# Execute the workflow
output = app.invoke(input_state)
print(output["result"])Key Components:
This setup allows you to define a clear workflow using LangGraph's |
Beta Was this translation helpful? Give feedback.
-
|
@dosu above is missing the plan and execute part. It doesn't have:
|
Beta Was this translation helpful? Give feedback.
-
|
To incorporate the Plan and Execute pattern using LangGraph with a Planner, Task Fetching Unit, and Joiner, you can structure your code as follows:
Here's a sample implementation: from langgraph.graph import StateGraph, START, END
from langgraph.checkpoint.memory import MemorySaver
from langchain_core.language_models import BaseChatModel
from langchain_openai import ChatOpenAI
from langchain_community.tools import SQLDatabase
from typing import Any, Dict, List
# Define the LLM
llm = ChatOpenAI(model="gpt-4-turbo-preview")
# Define the SQL Database connection
db = SQLDatabase.from_uri("postgresql://username:password@localhost/dbname")
# Define the SQL agent with necessary methods
class SQLAgent:
def parse_question(self, state: Dict[str, Any], memory: MemorySaver):
question = state["input"]
memory.save("parsed_question", question)
return {"parsed_question": question}
def generate_sql(self, state: Dict[str, Any], memory: MemorySaver):
parsed_question = memory.load("parsed_question")
sql_query = f"SELECT * FROM sales WHERE month = '{parsed_question}'"
memory.save("sql_query", sql_query)
return {"sql_query": sql_query}
def validate_sql(self, state: Dict[str, Any], memory: MemorySaver):
sql_query = memory.load("sql_query")
return {"validated_sql": sql_query}
def execute_sql(self, state: Dict[str, Any], memory: MemorySaver):
validated_sql = memory.load("validated_sql")
result = db.execute(validated_sql)
return {"result": result}
# Workflow manager to create the StateGraph
class WorkflowManager:
def __init__(self):
self.sql_agent = SQLAgent()
self.memory = MemorySaver()
def create_workflow(self) -> StateGraph:
workflow = StateGraph()
# Add nodes to the graph
workflow.add_node("parse_question", lambda state: self.sql_agent.parse_question(state, self.memory))
workflow.add_node("generate_sql", lambda state: self.sql_agent.generate_sql(state, self.memory))
workflow.add_node("validate_sql", lambda state: self.sql_agent.validate_sql(state, self.memory))
workflow.add_node("execute_sql", lambda state: self.sql_agent.execute_sql(state, self.memory))
# Define edges
workflow.add_edge(START, "parse_question")
workflow.add_edge("parse_question", "generate_sql")
workflow.add_edge("generate_sql", "validate_sql")
workflow.add_edge("validate_sql", "execute_sql")
workflow.add_edge("execute_sql", END)
return workflow
# Example usage
workflow_manager = WorkflowManager()
workflow = workflow_manager.create_workflow()
app = workflow.compile(checkpointer=workflow_manager.memory)
# Input state
input_state = {"input": "last month"}
# Execute the workflow
output = app.invoke(input_state)
print(output["result"])Key Components:
This setup allows you to define a clear workflow using LangGraph's |
Beta Was this translation helpful? Give feedback.
Uh oh!
There was an error while loading. Please reload this page.
Uh oh!
There was an error while loading. Please reload this page.
-
I am trying to execute the Langchain's LLM Compiler example code at this page.
When I run the below code, it hangs in
plan_and_schedule(state)andschedule_tasks.invoke()doesn't work or doesn't ends.Here's my code:
Here's my
output_parser_v2code:Here's my
math_toolscode:When I execute above, I don't see the code move forward from
schedule_tasks.invoke()and print output of this line.Beta Was this translation helpful? Give feedback.
All reactions