Three tools and a model that picks
The desk answers two kinds of question: where an order is, and what the shop's policies say. Three tools cover both, and each one is something you wrote earlier.
The orders carry an owner as well as a status, and Customer is the runtime context: the name comes from your code, never from the model.
from dataclasses import dataclass
from langchain.tools import ToolRuntime, tool
ORDERS = {"A17": ("ravi", "shipped on 3 March"), "C40": ("mei", "waiting for stock")}
@dataclass
class Customer:
name: str@tool
def lookup_order(order_id: str, runtime: ToolRuntime[Customer]) -> str:
"""Look up one of the customer's orders by its id, such as A17."""
owner, status = ORDERS.get(order_id, (None, None))
if owner != runtime.context.name:
return f"{order_id} is not one of your orders."
return f"{order_id} {status}."The lookup answers only about the asking customer's orders. A question about someone else's order gets the same reply as a question about an order that does not exist, which is the answer a shop should give.
@tool
def refund_order(order_id: str, runtime: ToolRuntime[Customer]) -> str:
"""Refund one of the customer's orders in full. This cannot be undone."""
owner, _ = ORDERS.get(order_id, (None, None))
if owner != runtime.context.name:
return f"{order_id} is not one of your orders, so it cannot be refunded."
return f"Refunded {order_id}."refund_order checks the owner too, so a customer cannot refund someone else's order even if a reviewer approves the call by mistake. The third tool is search_policies, the one over the vector store, unchanged.
A model that picks between them
An order id means the order tools; anything else is a question for the policies. When a tool has answered, its text is the reply.
import re
from langchain.messages import AIMessage
from shop_model import ShopModel
class DeskModel(ShopModel):
def decide(self, messages):
last = messages[-1]
if last.type == "tool" and last.text == "No policy covers this.":
return AIMessage("Our policies do not cover that. A person will reply.")
if last.type == "tool" or re.findall(r"\b[A-Z]\d+\b", last.text):
return super().decide(messages)
query = {"name": "search_policies", "args": {"query": last.text}, "id": "call_p"}
return AIMessage("", tool_calls=[query])from langchain.agents import create_agent
from desk_model import DeskModel
from search import search_policies
from tools import Customer, lookup_order, refund_order
desk = create_agent(DeskModel(), tools=[lookup_order, refund_order, search_policies],
context_schema=Customer)
for text in ["Where is A17?", "Is shipping free?", "Where is C40?"]:
result = desk.invoke({"messages": [{"role": "user", "content": text}]},
context=Customer("ravi"))
print(text, "->", result["messages"][-1].text)Three questions, three paths: an order Ravi owns, a policy found in the shipping document, and C40, which belongs to Mei. No rules yet, and nothing stopping a refund.
- Ask about an order id the shop has never heard of.
- Take
search_policiesout of the tool list and ask the shipping question again. - Print the whole message list for one question and count the steps.
Slow is fine. Stopping is the only problem.