Using openai.with_structured_output and tools together

[qubzes]

How can i use openai.with_structured_output and tools together.

i can do this with PydanticOutputParser, but I would like to use with structured output instead since it has better results

from typing import Any, Dict, List

from langchain_core.exceptions import OutputParserException
from langchain_core.messages import AIMessage, HumanMessage
from langchain_core.output_parsers import PydanticOutputParser
from langchain_core.prompts import ChatPromptTemplate, MessagesPlaceholder
from langgraph.checkpoint.memory import MemorySaver
from langgraph.graph import END, START, MessagesState, StateGraph
from langgraph.prebuilt import ToolNode

from app.schemas.chat import AssistantResponse
from app.services.prompt import (
BASE_PROMPT,
CONTACT_PROMPT,
SERVICE_PROMPT,
)
from app.services.tools import get_matching_services, search_for_contact
from app.services.utils import openai

def create_workflow(tools: List[Any], system_message: str):
"""Create a workflow using LCEL (LangChain Expression Language)"""
parser = PydanticOutputParser(pydantic_object=AssistantResponse)

def process_ai_response(parsed_response: AssistantResponse) -> dict:
    """Process AI response"""
    return {
        "messages": [
            AIMessage(
                content=parsed_response.message,
                additional_kwargs={
                    "quick_replies": parsed_response.quick_replies,
                    "smart_recommendations": parsed_response.smart_recommendations,
                },
            )
        ]
    }

def call_model(state: MessagesState):
    """Process the message through the model and parser"""
    format_instructions = parser.get_format_instructions()
    print(format_instructions)
    prompt = ChatPromptTemplate.from_messages(
        [
            ("system", system_message + f"\n{format_instructions}"),
            MessagesPlaceholder(variable_name="messages"),
        ]
    )
    chain = (
        prompt | openai.bind_tools(tools) | parser | process_ai_response
    )

    result = chain.invoke(
        {
            "messages": state["messages"],
        }
    )
    print(result)
    return result

if tools:
    tool_node = ToolNode(tools)

    def should_continue(state: MessagesState):
        """Determine if we should continue to tools or end"""
        last_message = state["messages"][-1]
        return "tools" if last_message.tool_calls else END

    workflow = StateGraph(MessagesState)
    workflow.add_node("agent", call_model)
    workflow.add_node("tools", tool_node)
    workflow.add_edge(START, "agent")
    workflow.add_conditional_edges("agent", should_continue)
    workflow.add_edge("tools", "agent")
else:
    # Create simple graph without tools
    workflow = StateGraph(MessagesState)
    workflow.add_node("agent", call_model)
    workflow.add_edge(START, "agent")
    workflow.add_edge("agent", END)

return workflow.compile(checkpointer=MemorySaver())

service_workflow = create_workflow(
tools=[get_matching_services], system_message=SERVICE_PROMPT
)

contact_workflow = create_workflow(
tools=[search_for_contact], system_message=CONTACT_PROMPT
)

general_workflow = create_workflow(tools=[], system_message=BASE_PROMPT)

def create_agent_thread(workflow):
"""Create a new conversation thread"""
checkpointer = MemorySaver()
thread_id = id(checkpointer)

def send_message(message: str) -> Dict[str, Any]:
    # Process message through workflow
    try:
        response = workflow.invoke(
            {"messages": [HumanMessage(content=message)]},
            config={"configurable": {"thread_id": thread_id}},
        )
        final_message = response.get("messages", [])[-1]

        return {
            "content": final_message.content,
            "quick_replies": final_message.additional_kwargs.get(
                "quick_replies", []
            ),
            "smart_recommendations": final_message.additional_kwargs.get(
                "smart_recommendations", []
            ),
        }
    except OutputParserException as e:
        print("Error", e)
        return {
            "content": "An unexpected error occurred. Please try again.",
            "quick_replies": [],
            "smart_recommendations": [],
        }

return {"thread_id": thread_id, "send_message": send_message}

[dosubot[bot]]

Hello, @ThePrimeJnr! I'm here to help you with any bugs, questions, or contributions you have. Let me know how I can assist you while you wait for a human maintainer.

You cannot directly combine with_structured_output and bind_tools in LangChain's ChatOpenAI. The issue arises because with_structured_output returns a RunnableSequence that doesn't support the bind_tools method. Instead, you should create separate LLM instances for different purposes: one for structured output and another for using tools ^[1].

To continue talking to Dosu, mention @dosu.

---

^{Help Dosu learn! Give it feedback: Great Response | Irrelevant Answer | Incorrect Sources | Too Verbose | Hallucination | Other | Bug Report}

[qubzes]

Thanks for youre response. if you needed to hve a struture output and still bing tools like in my case what would you do?
@dosu

[dosubot[bot]]

To achieve structured output while still binding tools in LangChain, you can use the with_structured_output method in combination with bind_tools. Here's how you can do it:

1. Define the Output Schema: Use a Pydantic class to define the structure of your output.

2. Bind Tools: Use the bind_tools method to bind any necessary tools to your model.

3. Use with_structured_output: Apply this method to ensure the output matches your defined schema.

Here's an example:

from langchain_community.chat_models import ChatLlamaCpp
from pydantic import BaseModel

class AnswerWithJustification(BaseModel):
    answer: str
    justification: str

llm = ChatLlamaCpp(
    temperature=0.,
    model_path="./SanctumAI-meta-llama-3-8b-instruct.Q8_0.gguf",
    n_ctx=10000,
    n_gpu_layers=4,
    n_batch=200,
    max_tokens=512,
    n_threads=multiprocessing.cpu_count() - 1,
    repeat_penalty=1.5,
    top_p=0.5,
    stop=["<|end_of_text|>", "<|eot_id|>"],
)

# Bind tools if necessary
tools = [...]  # Define your tools here
llm = llm.bind_tools(tools)

# Use with_structured_output to define the output schema
structured_llm = llm.with_structured_output(AnswerWithJustification)

# Invoke the model
result = structured_llm.invoke("What weighs more a pound of bricks or a pound of feathers")
print(result)

This approach allows you to maintain structured output while utilizing tool bindings by ensuring the model's response conforms to the desired structure ^[1][2].

To continue talking to Dosu, mention @dosu.

---

^{Help Dosu learn! Give it feedback: Great Response | Irrelevant Answer | Incorrect Sources | Too Verbose | Hallucination | Other | Bug Report}

[mtene]

To achieve structured output while still binding tools in LangChain, you can use the with_structured_output method in combination with bind_tools. Here's how you can do it:

1. Define the Output Schema: Use a Pydantic class to define the structure of your output.
2. Bind Tools: Use the bind_tools method to bind any necessary tools to your model.
3. Use with_structured_output: Apply this method to ensure the output matches your defined schema.

Here's an example:

from langchain_community.chat_models import ChatLlamaCpp
from pydantic import BaseModel

class AnswerWithJustification(BaseModel):
    answer: str
    justification: str

llm = ChatLlamaCpp(
    temperature=0.,
    model_path="./SanctumAI-meta-llama-3-8b-instruct.Q8_0.gguf",
    n_ctx=10000,
    n_gpu_layers=4,
    n_batch=200,
    max_tokens=512,
    n_threads=multiprocessing.cpu_count() - 1,
    repeat_penalty=1.5,
    top_p=0.5,
    stop=["<|end_of_text|>", "<|eot_id|>"],
)

# Bind tools if necessary
tools = [...]  # Define your tools here
llm = llm.bind_tools(tools)

# Use with_structured_output to define the output schema
structured_llm = llm.with_structured_output(AnswerWithJustification)

# Invoke the model
result = structured_llm.invoke("What weighs more a pound of bricks or a pound of feathers")
print(result)

This approach allows you to maintain structured output while utilizing tool bindings by ensuring the model's response conforms to the desired structure [1][2].

To continue talking to Dosu, mention @dosu.

Help Dosu learn! Give it feedback: Great Response | Irrelevant Answer | Incorrect Sources | Too Verbose | Hallucination | Other | Bug Report

@dosu this contradicts your initial answer: llm = llm.bind_tools(tools) returns a RunnableSequence which does not support the call llm.with_structured_output(AnswerWithJustification). It would be great, indeed, if the two could be chained, but that is not possible in langchain (see my comment below).

[mtene]

I think the need to combine tools and structured outputs on the same LLM instance is legitimate. Imagine a graph where the LLM is called in a loop until it produces no more tool call requests. The resulting structured output then goes to a validation node, which will fallback to the LLM loop for any errors or return the output to the caller if it satisfies all pydantic constraints. This allows integration of GenAI as a reliable component in a programmatic workflow, with built-in error handling. This simple structure can be upgraded with reasoning, planning or other patterns as needed, yet the tool calling + structured output LLM remains the central building block.

The langchain design choice to disallow this is unfortunate, despite e.g. OpenAI LLMs explicitly allowing the specification of both tool definitions and structured output schema in the same request (and making use of one or the other in each turn). The workarounds that come to mind suffer from limitations:

• two LLM instances, one for tool calls (tool_choice=auto) and another to structure the output (tool_choice=none) - the output specification should drive decisions on what tools to call. Including it in the system prompt of the tool calling LLM might help, but it is a soft suggestion rather than a hard constraint. Since the LLM output is to be consumed programmatically, there is no wiggle room. Also, when the user query is simple enough to be answered without any tool calls, one still needs 2 LLM calls, impacting latency and doubling the number of tokens (aka the cost of 'Hi').
• one LLM instance with an additional "sink" tool for the output (likely tool_choice=required) - all information is available and the response schema is enforced, but the model may not honor the tool calling precedence and finalize too early. Also parallel_tool_calls=true cannot be used as it would be ambiguous if the sink tool appears among them. Finally, it is awkward to retrieve the response via a tool call e.g. under a langgraph ToolNode, as that will be far away from the consumption point in the call stack and execution may continue to other nodes afterwards.

@dosu any thoughts on the above?