Агенты каталога и счетов LangGraph с возможностью отображения в LangSmith
Каталог и инвойс-агенты в LangGraph: маршрутизация руководителя, прерывания со стороны человека, долгосрочная память и циклы оценки LangSmith, обеспечивающие точность.
Источники: Building a Multi-Agent AI System with LangGraph and LangSmith
LangSmith существует для обеспечения возможности наблюдения за многокомпонентными системами: записи действий, результаты оценки и запуски наборов данных показывают, что видел и возвращал каждый узел. В этой версии специалист по каталогам и специалист по счетам-фактурам связаны через структуру LangGraph в стиле руководителя, после чего добавляются механизмы участия человека и долгосрочная память.
Цель LangSmith
Используйте его в качестве инструмента для наблюдения и оценки приложений на основе больших языковых моделей, а не в качестве самого оркестратора. Возможность отладки помогает сохранять прозрачность в сложных многокомпонентных системах.
Архитектура многокомпонентных систем
Родительская структура направляет запросы к подагентам, отвечающим за каталоги или счета-фактуры, причем каждый из них имеет инструменты и указания, адаптированные под соответствующую область. Общее состояние используется для передачи сообщений, идентификаторов клиентов и флагов для прерываний.
import urllib.request
from langchain_community.utilities.sql_database import SQLDatabase
from sqlalchemy import create_engine
def get_northwind_db():
db_path = "/tmp/northwind.db"
url = "https://raw.githubusercontent.com/jpwhite3/northwind-SQLite3/main/dist/northwind.db"
urllib.request.urlretrieve(url, db_path)
engine = create_engine(f"sqlite:///{db_path}")
return SQLDatabase(engine)
db = get_northwind_db()
Подключите этот фрагмент к структуре перед продолжением работы; неполные структуры скрывают ошибки маршрутизации до дня демонстрации.
import os
os.environ["OPENAI_API_KEY"] = "your-openai-api-key"
os.environ["LANGSMITH_API_KEY"] = "your-langsmith-api-key"
os.environ["LANGSMITH_TRACING"] = "true"
os.environ["LANGSMITH_PROJECT"] = "your-langsmith-project"
from langsmith import utils
print(f"LangSmith tracing is enabled: {utils.tracing_is_enabled()}")
По возможности протестируйте эту единицу изолированно, затем соберите её. Изоляция позволяет рано обнаружить ошибки в схеме инструмента.
from langchain_openai import ChatOpenAI
from langgraph.checkpoint.memory import MemorySaver
from langgraph.store.memory import InMemoryStore
llm = ChatOpenAI(model="gpt-4o-mini", temperature=0)
# Long-term memory: persists user preferences across conversations
in_memory_store = InMemoryStore()
# Short-term memory (checkpointer): maintains context within a single session
checkpointer = MemorySaver()
Приведённый выше пункт является конкретным шагом в процессе сборки — сохраняйте стабильность названий, чтобы трассировки LangSmith оставались сопоставимыми между версиями.
from typing import Annotated, List
from typing_extensions import TypedDict
from langgraph.graph.message import AnyMessage, add_messages
from langgraph.managed.is_last_step import RemainingSteps
class State(TypedDict):
"""State schema for the multi-agent customer support workflow."""
# Verified 5-letter Northwind customer code (e.g. "ALFKI")
customer_id: str
# Conversation history with automatic message aggregation
messages: Annotated[List[AnyMessage], add_messages]
# User preferences and context loaded from long-term memory store
loaded_memory: str
# Counter to prevent infinite recursion in agent workflow
remaining_steps: RemainingSteps
Подключите этот фрагмент к графу перед тем, как переходить дальше; частичные графы скрывают ошибки маршрутизации до дня демонстрации.
import ast
from langchain_core.tools import tool
@tool
def search_products_by_name(name: str) -> str:
"""
Search for products by name in the Northwind catalog.
Args:
name (str): Full or partial product name to search for.
Returns:
str: Products with category, price, and stock level.
"""
return db.run(
f"""
SELECT p.ProductName, c.CategoryName, p.UnitPrice, p.UnitsInStock
FROM Products p
JOIN Categories c ON c.CategoryID = p.CategoryID
WHERE p.ProductName LIKE '%{name}%'
ORDER BY p.ProductName
LIMIT 10;
""",
include_columns=True
)
@tool
def search_products_by_category(category: str) -> str:
"""
Look up all products in a specific category.
Available categories: Beverages, Condiments, Confections, Dairy Products,
Grains/Cereals, Meat/Poultry, Produce, Seafood.
Args:
category (str): Category name, partial match allowed.
Returns:
str: Products with price and stock level.
"""
return db.run(
f"""
SELECT p.ProductName, c.CategoryName, p.UnitPrice, p.UnitsInStock
FROM Products p
JOIN Categories c ON c.CategoryID = p.CategoryID
WHERE c.CategoryName LIKE '%{category}%'
ORDER BY p.ProductName;
""",
include_columns=True
)
@tool
def check_product_stock(product_name: str) -> str:
"""
Check whether a specific product is currently in stock.
Args:
product_name (str): The product name to check.
Returns:
str: Stock level and whether the product is discontinued.
"""
return db.run(
f"""
SELECT ProductName, UnitsInStock, Discontinued
FROM Products
WHERE ProductName LIKE '%{product_name}%';
""",
include_columns=True
)
catalog_tools = [search_products_by_name, search_products_by_category, check_product_stock]
llm_with_catalog_tools = llm.bind_tools(catalog_tools)
По возможности протестируйте эту единицу изолированно, затем соберите её. Изоляция позволяет рано обнаружить ошибки в схеме инструмента.
from langgraph.prebuilt import ToolNode
# ToolNode executes whichever tool the LLM called
catalog_tool_node = ToolNode(catalog_tools)
Приведённый выше пункт является конкретным шагом в процессе сборки — сохраняйте стабильность названий, чтобы трассировки LangSmith оставались сопоставимыми между версиями.
from langchain_core.messages import SystemMessage, HumanMessage
from langchain_core.runnables import RunnableConfig
def generate_catalog_assistant_prompt(memory: str = "None") -> str:
"""Generate the system prompt, injecting any saved customer preferences."""
return f"""You are a product catalog specialist for Northwind Traders,
a specialty foods distributor.
You help customers find products by name or category using your tools.
Available categories: Beverages, Condiments, Confections, Dairy Products,
Grains/Cereals, Meat/Poultry, Produce, Seafood.
CORE RESPONSIBILITIES:
- Always search before answering — never invent a product, price, or stock level
- If nothing matches, say so and suggest a broader search term
- Include price and stock status when listing products
SEARCH GUIDELINES:
1. Always perform a search before concluding something is unavailable
2. Try partial name matches if exact matches fail
3. If exact matches aren't found, try alternative spellings
Prior saved customer preferences: {memory}
Message history is also attached.
"""
def catalog_assistant(state: State, config: RunnableConfig):
"""
Catalog assistant node — the LLM reasoning step.
Reads the conversation and loaded memory, then either calls a catalog
tool or generates a final response.
"""
memory = "None"
if "loaded_memory" in state:
memory = state["loaded_memory"]
prompt = generate_catalog_assistant_prompt(memory)
response = llm_with_catalog_tools.invoke(
[SystemMessage(content=prompt)] + state["messages"]
)
return {"messages": [response]}
Подключите этот фрагмент к графу перед тем, как переходить дальше; частичные графы скрывают ошибки маршрутизации до дня демонстрации.
from langgraph.graph import StateGraph, START, END
def should_continue(state: State, config: RunnableConfig):
"""
Conditional edge: checks the last message for tool_calls.
If tool calls exist, route to the tool node. Otherwise end.
"""
last_message = state["messages"][-1]
if not last_message.tool_calls:
return "end"
else:
return "continue"
По возможности протестируйте эту единицу изолированно, затем соберите её. Изоляция позволяет рано обнаружить ошибки в схеме инструмента.
catalog_workflow = StateGraph(State)
catalog_workflow.add_node("catalog_assistant", catalog_assistant)
catalog_workflow.add_node("catalog_tool_node", catalog_tool_node)
catalog_workflow.add_edge(START, "catalog_assistant")
catalog_workflow.add_conditional_edges(
"catalog_assistant",
should_continue,
{
"continue": "catalog_tool_node",
"end": END,
}
)
# After tool execution, loop back to the assistant to process the result
catalog_workflow.add_edge("catalog_tool_node", "catalog_assistant")
product_catalog_subagent = catalog_workflow.compile(
name="product_catalog_subagent",
checkpointer=checkpointer,
store=in_memory_store
)
Приведённый выше пункт является конкретным шагом в процессе сборки — сохраняйте стабильность названий, чтобы трассировки LangSmith оставались сопоставимыми между версиями.
import uuid
thread_id = str(uuid.uuid4())
config = {"configurable": {"thread_id": thread_id}}
question = "Do you have any Seafood products? What are the prices?"
result = product_catalog_subagent.invoke(
{"messages": [HumanMessage(content=question)]},
config=config
)
for message in result["messages"]:
message.pretty_print()
Подключите этот фрагмент к графу перед тем, как переходить дальше; частичные графы скрывают ошибки маршрутизации до дня демонстрации.
@tool
def get_invoices_by_customer_sorted_by_date(customer_id: str) -> str:
"""
Retrieve all orders for a customer sorted by date, most recent first.
Args:
customer_id (str): The 5-letter Northwind customer code (e.g. "ALFKI").
Returns:
str: Orders with order ID, date, and shipping info.
"""
return db.run(
f"""
SELECT OrderID, OrderDate, ShippedDate, ShipCity, ShipCountry
FROM Orders
WHERE CustomerID = '{customer_id}'
ORDER BY OrderDate DESC
LIMIT 10;
""",
include_columns=True
)
@tool
def get_invoices_sorted_by_unit_price(order_id: str) -> str:
"""
Retrieve the line items of a specific order, highest unit price first.
Args:
order_id (str): The order ID number.
Returns:
str: Line items with product name, unit price, quantity, and discount.
"""
return db.run(
f"""
SELECT p.ProductName, od.UnitPrice, od.Quantity, od.Discount
FROM "Order Details" od
JOIN Products p ON p.ProductID = od.ProductID
WHERE od.OrderID = {order_id}
ORDER BY od.UnitPrice DESC;
""",
include_columns=True
)
@tool
def get_employee_by_invoice_and_customer(order_id: str, customer_id: str) -> str:
"""
Find the Northwind employee (sales rep) who handled a specific order.
Args:
order_id (str): The order ID.
customer_id (str): The 5-letter customer code.
Returns:
str: Employee name and title.
"""
return db.run(
f"""
SELECT e.FirstName, e.LastName, e.Title
FROM Employees e
JOIN Orders o ON o.EmployeeID = e.EmployeeID
WHERE o.OrderID = {order_id}
AND o.CustomerID = '{customer_id}';
""",
include_columns=True
)
invoice_tools = [
get_invoices_by_customer_sorted_by_date,
get_invoices_sorted_by_unit_price,
get_employee_by_invoice_and_customer,
]
По возможности протестируйте эту единицу изолированно, затем соберите её. Изоляция позволяет рано обнаружить ошибки в схеме инструмента.
from langgraph.prebuilt import create_react_agent
invoice_subagent_prompt = """You are a subagent among a team of assistants.
You are specialized for retrieving order and invoice information for Northwind Traders customers.
You have access to three tools:
- get_invoices_by_customer_sorted_by_date: retrieves all orders for a customer, most recent first
- get_invoices_sorted_by_unit_price: retrieves the items in a specific order
- get_employee_by_invoice_and_customer: finds the sales rep who handled a specific order
CORE RESPONSIBILITIES:
- Retrieve and present order information accurately from the database
- Provide details about orders including items, prices, and shipping
- Always maintain a professional, friendly, and patient tone
"""
invoice_information_subagent = create_react_agent(
llm,
tools=invoice_tools,
name="invoice_information_subagent",
prompt=invoice_subagent_prompt,
state_schema=State,
checkpointer=checkpointer,
store=in_memory_store
)
Приведённый выше пункт является конкретным шагом в процессе сборки — сохраняйте стабильность названий, чтобы трассировки LangSmith оставались сопоставимыми между версиями.
thread_id = str(uuid.uuid4())
config = {"configurable": {"thread_id": thread_id}}
question = "My customer id is ALFKI. What was my most recent order, and who was the employee that helped me with it?"
result = invoice_information_subagent.invoke(
{"messages": [HumanMessage(content=question)]},
config=config
)
for message in result["messages"]:
message.pretty_print()
Подключите этот фрагмент к графу перед тем, как переходить дальше; частичные графы скрывают ошибки маршрутизации до дня демонстрации.
from langgraph_supervisor import create_supervisor
supervisor_prompt = """You are an expert customer support assistant for Northwind Traders,
a specialty foods distributor.
You are dedicated to providing exceptional service and ensuring customer queries are resolved.
You have a team of two subagents that you can use to help answer customer queries:
1. product_catalog_subagent: handles questions about products, categories, prices, and stock availability.
This agent has access to the Northwind product catalog.
2. invoice_information_subagent: handles questions about a customer's past orders,
order line items, shipping, and their assigned sales representative.
Based on the customer's message, route it to the correct specialist.
This could be one step in an inquiry that needs multiple sub-agent calls.
"""
supervisor_prebuilt_workflow = create_supervisor(
agents=[invoice_information_subagent, product_catalog_subagent],
output_mode="last_message",
model=llm,
prompt=(supervisor_prompt),
state_schema=State
)
supervisor_prebuilt = supervisor_prebuilt_workflow.compile(
name="product_catalog_subagent",
checkpointer=checkpointer,
store=in_memory_store
)
По возможности протестируйте эту единицу изолированно, затем соберите её. Изоляция позволяет рано обнаружить ошибки в схеме инструмента.
thread_id = str(uuid.uuid4())
config = {"configurable": {"thread_id": thread_id}}
result = supervisor_prebuilt.invoke(
{"messages": [HumanMessage(content="What Beverages do you have in stock?")],
"customer_id": "ALFKI"},
config=config
)
for message in result["messages"]:
message.pretty_print()
Приведённый выше пункт является конкретным шагом в процессе сборки — сохраняйте стабильность названий, чтобы трассировки LangSmith оставались сопоставимыми между версиями.
from pydantic import BaseModel, Field
from typing import Optional
from langgraph.types import interrupt
class UserInput(BaseModel):
"""Schema for parsing the customer's account identifier."""
identifier: str = Field(
description="The customer identifier extracted from the message. "
"Can be a 5-letter customer ID (e.g. ALFKI), phone number, "
"or company name. Return empty string if not found."
)
# Structured LLM — always returns a UserInput object, never free text
structured_llm = llm.with_structured_output(schema=UserInput)
structured_system_prompt = """You are a customer service representative responsible for
verifying customer identity at Northwind Traders.
Only extract the customer's account identifier from the message.
This can be a 5-letter customer ID (like ALFKI), a phone number, or a company name.
If the customer has not provided any identifier, return an empty string.
"""
Подключите этот фрагмент к графу перед тем, как переходить дальше; частичные графы скрывают ошибки маршрутизации до дня демонстрации.
def get_customer_id_from_identifier(identifier: str) -> Optional[str]:
"""
Retrieve a CustomerID from the Northwind database using one of three identifiers:
1. Direct 5-letter customer ID (e.g. "ALFKI")
2. Phone number
3. Company name (partial match)
"""
if not identifier:
return None
identifier = identifier.strip()
# Direct customer ID: 5 alpha characters
if identifier.replace(" ", "").isalpha() and len(identifier.replace(" ", "")) == 5:
result = db.run(
f"SELECT CustomerID FROM Customers WHERE CustomerID = '{identifier.upper()}';"
)
if result and result != "[]":
try:
formatted = ast.literal_eval(result)
if formatted:
return formatted[0][0]
except Exception:
pass
# Phone number: contains digits
if any(c.isdigit() for c in identifier):
result = db.run(
f"SELECT CustomerID FROM Customers WHERE Phone = '{identifier}';"
)
if result and result != "[]":
try:
formatted = ast.literal_eval(result)
if formatted:
return formatted[0][0]
except Exception:
pass
# Company name: partial match
result = db.run(
f"SELECT CustomerID FROM Customers WHERE CompanyName LIKE '%{identifier}%' LIMIT 1;"
)
if result and result != "[]":
try:
formatted = ast.literal_eval(result)
if formatted:
return formatted[0][0]
except Exception:
pass
return None
По возможности протестируйте эту единицу изолированно, затем соберите её. Изоляция позволяет рано обнаружить ошибки в схеме инструмента.
def verify_info(state: State, config: RunnableConfig):
"""
Verifies the customer's identity.
- If customer_id is already in state: skip (already verified)
- If identifier found and valid in DB: set customer_id and confirm
- If not found: ask the customer politely for their details
"""
if state.get("customer_id") is not None:
return
system_instructions = """You are a Northwind Traders customer support agent
trying to verify the customer's identity as the first step of the support process.
Only after their account is verified will you be able to help them with orders.
To verify their identity, ask for their 5-letter Customer ID (e.g. ALFKI),
phone number, or company name."""
user_input = state["messages"][-1]
parsed_info = structured_llm.invoke(
[SystemMessage(content=structured_system_prompt), user_input]
)
identifier = parsed_info.identifier
customer_id = ""
if identifier:
customer_id = get_customer_id_from_identifier(identifier)
if customer_id:
intent_message = SystemMessage(
content=f"Thank you for providing your information! "
f"I was able to verify your account (Customer ID: {customer_id}). "
f"How can I help you today?"
)
return {
"customer_id": customer_id,
"messages": [intent_message]
}
else:
response = llm.invoke(
[SystemMessage(content=system_instructions)] + state["messages"]
)
return {"messages": [response]}
def human_input(state: State, config: RunnableConfig):
"""
Human-in-the-loop node. Pauses the graph using interrupt() and waits
for the user to provide their customer identifier.
"""
user_input = interrupt("Please provide your Customer ID, phone number, or company name.")
return {"messages": [user_input]}
def should_interrupt(state: State, config: RunnableConfig):
"""Routes to human_input if customer_id is not verified, else continues."""
if state.get("customer_id") is not None:
return "continue"
else:
return "interrupt"
Приведённый выше пункт является конкретным шагом в процессе сборки — сохраняйте стабильность названий, чтобы трассировки LangSmith оставались сопоставимыми между версиями.
multi_agent_verify = StateGraph(State)
multi_agent_verify.add_node("verify_info", verify_info)
multi_agent_verify.add_node("human_input", human_input)
multi_agent_verify.add_node("supervisor", supervisor_prebuilt)
multi_agent_verify.add_edge(START, "verify_info")
multi_agent_verify.add_conditional_edges(
"verify_info",
should_interrupt,
{
"continue": "supervisor",
"interrupt": "human_input",
}
)
multi_agent_verify.add_edge("human_input", "verify_info")
multi_agent_verify.add_edge("supervisor", END)
multi_agent_verify_graph = multi_agent_verify.compile(
name="multi_agent_verify",
checkpointer=checkpointer,
store=in_memory_store
)
Подключите этот фрагмент к графу перед тем, как переходить дальше; частичные графы скрывают ошибки маршрутизации до дня демонстрации.
from langgraph.types import Command
thread_id = str(uuid.uuid4())
config = {"configurable": {"thread_id": thread_id}}
result = multi_agent_verify_graph.invoke(
{"messages": [HumanMessage(content="What was in my last order?")]},
config=config
)
По возможности протестируйте эту единицу изолированно, затем соберите её. Изоляция позволяет рано обнаружить ошибки в схеме инструмента.
result = multi_agent_verify_graph.invoke(
Command(resume="My customer ID is ALFKI"),
config=config
)
Приведённый выше пункт является конкретным шагом в процессе сборки — сохраняйте стабильность названий, чтобы трассировки LangSmith оставались сопоставимыми между версиями.
result = multi_agent_verify_graph.invoke(
{"messages": [HumanMessage(content="What Beverages do you carry?")]},
config=config
)
Подключите этот фрагмент к графу перед тем, как переходить дальше; частичные графы скрывают ошибки маршрутизации до дня демонстрации.
from langgraph.store.base import BaseStore
def format_user_memory(user_data: dict) -> str:
"""Formats the stored UserProfile into a string for injection into prompts."""
profile = user_data.get("memory")
result = ""
if hasattr(profile, "product_preferences") and profile.product_preferences:
result += f"Product Category Preferences: {', '.join(profile.product_preferences)}"
return result.strip()
def load_memory(state: State, config: RunnableConfig, store: BaseStore):
"""
Loads the customer's saved product preferences from the long-term memory store.
Runs right after verification and before the supervisor handles the query.
"""
user_id = config["configurable"].get("user_id", state["customer_id"])
namespace = ("memory_profile", user_id)
key = "user_memory"
existing_memory = store.get(namespace, key)
formatted_memory = ""
if existing_memory and existing_memory.value:
formatted_memory = format_user_memory(existing_memory.value)
return {"loaded_memory": formatted_memory}
По возможности протестируйте эту единицу изолированно, затем соберите её. Изоляция позволяет рано обнаружить ошибки в схеме инструмента.
class UserProfile(BaseModel):
"""Structured profile saved to long-term memory after each conversation."""
customer_id: str = Field(description="The customer's ID code")
product_preferences: List[str] = Field(
description="List of product categories the customer expressed interest in, "
"e.g. ['Seafood', 'Beverages']"
)
Приведённый выше пункт является конкретным шагом в процессе сборки — сохраняйте стабильность названий, чтобы трассировки LangSmith оставались сопоставимыми между версиями.
create_memory_prompt = """You are an expert analyst observing a customer support conversation
at Northwind Traders.
You are tasked with analyzing the conversation and updating the customer's memory profile
with any newly identified product category preferences.
*IMPORTANT INFORMATION BELOW*
The conversation between the customer and the support agent:
{conversation}
The existing memory profile associated with the customer:
{memory_profile}
For each field, if there is no new information, do not update the value.
Take a deep breath and think carefully before responding.
"""
def create_memory(state: State, config: RunnableConfig, store: BaseStore):
"""
Analyzes the conversation and updates the customer's long-term memory profile
with any new product category preferences they expressed.
"""
user_id = str(config["configurable"].get("user_id", state["customer_id"]))
namespace = ("memory_profile", user_id)
key = "user_memory"
existing_memory = store.get(namespace, key)
formatted_memory = ""
if existing_memory and existing_memory.value:
formatted_memory = format_user_memory(existing_memory.value)
formatted_system_message = SystemMessage(
content=create_memory_prompt.format(
conversation=state["messages"],
memory_profile=formatted_memory
)
)
updated_memory = llm.with_structured_output(UserProfile).invoke(
[formatted_system_message]
)
store.put(namespace, key, {"memory": updated_memory})
Подключите этот фрагмент к графу перед тем, как переходить дальше; частичные графы скрывают ошибки маршрутизации до дня демонстрации.
multi_agent_final = StateGraph(State)
multi_agent_final.add_node("verify_info", verify_info)
multi_agent_final.add_node("human_input", human_input)
multi_agent_final.add_node("load_memory", load_memory)
multi_agent_final.add_node("supervisor", supervisor_prebuilt)
multi_agent_final.add_node("create_memory", create_memory)
multi_agent_final.add_edge(START, "verify_info")
multi_agent_final.add_conditional_edges(
"verify_info",
should_interrupt,
{
"continue": "load_memory",
"interrupt": "human_input",
}
)
multi_agent_final.add_edge("human_input", "verify_info")
multi_agent_final.add_edge("load_memory", "supervisor")
multi_agent_final.add_edge("supervisor", "create_memory")
multi_agent_final.add_edge("create_memory", END)
multi_agent_final_graph = multi_agent_final.compile(
name="multi_agent_verify",
checkpointer=checkpointer,
store=in_memory_store
)
По возможности протестируйте эту единицу изолированно, затем соберите её. Изоляция позволяет рано обнаружить ошибки в схеме инструмента.
thread_id = str(uuid.uuid4())
config = {"configurable": {"thread_id": thread_id}}
result = multi_agent_final_graph.invoke(
{"messages": [HumanMessage(
content="My customer ID is ALFKI. I am really interested in your Seafood and Beverages range!"
)]},
config=config
)
# Check what memory was saved
memory_key = in_memory_store.get(("memory_profile", "ALFKI"), "user_memory")
if memory_key:
print(f"Memory saved: {memory_key.value['memory'].product_preferences}")
Приведённый выше шаг является конкретным этапом сборки — сохраняйте стабильные имена, чтобы трассировка LangSmith оставалась сопоставимой между версиями.
thread_id_2 = str(uuid.uuid4())
config_2 = {"configurable": {"thread_id": thread_id_2}}
result2 = multi_agent_final_graph.invoke(
{"messages": [HumanMessage(
content="Hi, I'm back. Customer ID ALFKI. What would you recommend for me today?"
)]},
config=config_2
)
Подключите этот фрагмент к графу перед переходом дальше; частичные графы скрывают ошибки маршрутизации до дня демонстрации.
from langsmith import Client
from langsmith.evaluation import evaluate
client = Client()
examples = [
{
"input": {"question": "What Beverages do you carry?", "customer_id": "ALFKI"},
"output": "Chai"
},
{
"input": {"question": "What was in my most recent order?", "customer_id": "ALFKI"},
"output": "order"
},
{
"input": {"question": "I'd like a refund.", "customer_id": "ANATR"},
"output": "additional information"
},
{
"input": {
"question": "My name is Maria Anders. What is the order ID of my most recent purchase?",
"customer_id": None
},
"output": "verify"
}
]
dataset = client.create_dataset(
"northwind-agent-eval",
description="Test cases for the Northwind multi-agent support system",
)
for ex in examples:
client.create_example(
inputs=ex["input"],
outputs={"answer": ex["output"]},
dataset_id=dataset.id,
)
def run_agent(inputs: dict) -> dict:
thread_id = str(uuid.uuid4())
config = {"configurable": {"thread_id": thread_id}}
initial_state = {"messages": [HumanMessage(content=inputs["question"])]}
if inputs.get("customer_id"):
initial_state["customer_id"] = inputs["customer_id"]
result = multi_agent_final_graph.invoke(initial_state, config=config)
final_answer = next(
(m.content for m in reversed(result["messages"])
if hasattr(m, "content") and m.content), ""
)
return {"answer": final_answer}
results = evaluate(
run_agent,
data="northwind-agent-eval",
evaluators=["correctness"],
experiment_prefix="northwind-run-1",
)
По возможности протестируйте эту единицу изолированно, затем соберите её. Изоляция позволяет рано обнаружить ошибки в схеме инструмента.
Кратковременная и долговременная память
Точки контроля охватывают одну нить выполнения. Долговременное хранилище сохраняет предпочтения между нитями. Не используйте параметры TTL для предпочтений, как это делается с записями CDN.
Участие человека
Прервите работу, если отсутствует идентификатор; возобновите с идентификатором клиента; продолжайте ту же нить для последующих действий. Без точки контроля при возобновлении работы создаётся вымышленная история.
Оценка
Создайте небольшой «золотой» набор: каталог запросов, счета-фактуры и ловушки междоменного взаимодействия. Оцените точность маршрутизации и достоверность окончательных ответов в данных наборов LangSmith.
Заключение
Многоклиентская архитектура полезна, когда домены сильно отличаются по инструментам и правилам работы. В противном случае более экономичным и проще для оценки является один хорошо оснащенный агент с четкими механизмами прерывания.
Записывайте ответственных за принятие решений рядом с узлами и целями webhook-сообщений.
В ненадежных циклах предпочитайте запросы на проверку, содержащие только оператор SELECT.
Сохраняйте копии «золотых» инцидентов до изменения параметров сортировки сообщений.
Уведомляйте о циклах с неудачной проверкой так же, как сервисы уведомляют о сетевых ошибках типа 5xx.
С самого начала разделяйте клиентов с помощью меток в графе.
Документируйте результаты работы Terraform, которые использует агент во время выполнения.
Записывайте ответственных за принятие решений рядом с узлами и целями webhook-сообщений.
В ненадежных циклах предпочитайте запросы на проверку, содержащие только оператор SELECT.
Сделайте снимок критических инцидентов до изменения параметров триажа.
Уведомляйте при циклах с ошибками проверки так же, как сервисы уведомляют при кодах ошибок 5xx.
С самого начала разделяйте клиентских арендаторов с помощью меток в графе.
Задокументируйте результаты работы Terraform, которые использует фоновый процесс агента.
Укажите ответственных за принятие решений рядом с узлами и целями webhook-сообщений.
В ненадежных циклах предпочитайте запросы проверки, содержащие только оператор SELECT.
Сделайте снимок критических инцидентов до изменения параметров триажа.
Уведомляйте при циклах с ошибками проверки так же, как сервисы уведомляют при кодах ошибок 5xx.
С самого начала разделяйте клиентских арендаторов с помощью меток в графе.
Задокументируйте результаты работы Terraform, которые использует фоновый процесс агента.
Укажите ответственных за принятие решений рядом с узлами и целями webhook-сообщений.
В ненадежных циклах предпочитайте запросы проверки, содержащие только оператор SELECT.
Сделайте снимок критических инцидентов до изменения параметров триажа.
Сигнализируйте о циклах с неудачной верификацией так же, как сервисы сигнализируют о кодах ошибок 5xx.
С самого начала разделяйте клиентских абонентов с помощью меток в графе.
Документируйте результаты работы Terraform, которые использует движок агента.
Указывайте ответственных за принятие решений рядом с узлами и целями webhook-сообщений.
В ненадежных циклах предпочитайте запросы верификации, содержащие только оператор SELECT.
Создавайте снимки критических инцидентов перед изменением параметров сортировки.
Сигнализируйте о циклах с неудачной верификацией так же, как сервисы сигнализируют о кодах ошибок 5xx.
С самого начала разделяйте клиентских абонентов с помощью меток в графе.
Документируйте результаты работы Terraform, которые использует движок агента.
Указывайте ответственных за принятие решений рядом с узлами и целями webhook-сообщений.
В ненадежных циклах предпочитайте запросы верификации, содержащие только оператор SELECT.
Создавайте снимки критических инцидентов перед изменением параметров сортировки.
Сигнализируйте о циклах с неудачной верификацией так же, как сервисы сигнализируют о кодах ошибок 5xx.
С самого начала разделяйте клиентских арендаторов с помощью меток графа.
Документируйте результаты работы Terraform, которые использует движок агента.
Фиксируйте ответственных за принятие решений рядом с узлами и целями webhook-сообщений.
В ненадежных циклах предпочитайте запросы для проверки, содержащие только оператор SELECT.
Создавайте снимки критических инцидентов до изменения параметров триажа.
Уведомляйте о циклах с неудачной проверкой так же, как сервисы уведомляют о кодах ошибок 5xx.
С самого начала разделяйте клиентских арендаторов с помощью меток графа.
Документируйте результаты работы Terraform, которые использует движок агента.
Фиксируйте ответственных за принятие решений рядом с узлами и целями webhook-сообщений.
В ненадежных циклах предпочитайте запросы для проверки, содержащие только оператор SELECT.
Создавайте снимки критических инцидентов до изменения параметров триажа.
Уведомляйте о циклах с неудачной проверкой так же, как сервисы уведомляют о кодах ошибок 5xx.
С самого начала разделяйте клиентских арендаторов с помощью меток графа.
Задокументируйте выходные данные Terraform, которые использует движок агента.
Записывайте ответственных за принятие решений рядом с узлами и целями webhook.
В ненадежных циклах предпочитайте запросы проверки, содержащие только оператор SELECT.
Сделайте снимок критических инцидентов до изменения параметров триажа.
Подавайте уведомления при циклах с неудачной проверкой так же, как сервисы подают уведомления при ошибках 5xx.
С самого начала разделяйте тенанты клиентов с помощью меток графа.
Задокументируйте выходные данные Terraform, которые использует движок агента.
Записывайте ответственных за принятие решений рядом с узлами и целями webhook.
В ненадежных циклах предпочитайте запросы проверки, содержащие только оператор SELECT.
Сделайте снимок критических инцидентов до изменения параметров триажа.
Подавайте уведомления при циклах с неудачной проверкой так же, как сервисы подают уведомления при ошибках 5xx.
С самого начала разделяйте тенанты клиентов с помощью меток графа.
Задокументируйте выходные данные Terraform, которые использует движок агента.
Храните информацию об ответственных лицах, настройках и записях о возврате к предыдущему состоянию рядом с каждой диаграммой или изменением агента, чтобы демонстрации не превосходили возможности системы.
Храните информацию об ответственных лицах, настройках и записях о возврате к предыдущему состоянию рядом с каждой диаграммой или изменением агента, чтобы демонстрации не превосходили возможности системы.
Храните информацию об ответственных лицах, настройках и записях о возврате к предыдущему состоянию рядом с каждой диаграммой или изменением агента, чтобы демонстрации не превосходили возможности системы.
Храните информацию об ответственных лицах, настройках и записях о возврате к предыдущему состоянию рядом с каждой диаграммой или изменением агента, чтобы демонстрации не превосходили возможности системы.
Храните информацию об ответственных лицах, настройках и записях о возврате к предыдущему состоянию рядом с каждой диаграммой или изменением агента, чтобы демонстрации не превосходили возможности системы.
Храните информацию об ответственных лицах, настройках и записях о возврате к предыдущему состоянию рядом с каждой диаграммой или изменением агента, чтобы демонстрации не превосходили возможности системы.
Храните информацию об ответственных лицах, настройках и записях о возврате к предыдущему состоянию рядом с каждой диаграммой или изменением агента, чтобы демонстрации не превосходили возможности системы.
Храните информацию об ответственных лицах, настройках и записях о возврате к предыдущему состоянию рядом с каждой диаграммой или изменением агента, чтобы демонстрации не превосходили возможности системы.
Храните информацию об ответственных лицах, настройках и записях о возврате к предыдущему состоянию рядом с каждой диаграммой или изменением агента, чтобы демонстрации не превосходили возможности системы.
Храните информацию об ответственных лицах, настройках и записях о возврате к предыдущему состоянию рядом с каждой диаграммой или изменением агента, чтобы демонстрации не превосходили возможности системы.
Храните информацию об ответственных лицах, настройках и записях о возврате к предыдущему состоянию рядом с каждой диаграммой или изменением агента, чтобы демонстрации не превосходили возможности системы.
Храните информацию об ответственных лицах, настройках и записях о возврате к предыдущему состоянию рядом с каждой диаграммой или изменением агента, чтобы демонстрации не превосходили возможности системы.
Храните информацию об ответственных лицах, настройках и записях о возврате к предыдущему состоянию рядом с каждой диаграммой или изменением агента, чтобы демонстрации не превосходили возможности системы.
Храните информацию об ответственных лицах, настройках и записях о возврате к предыдущему состоянию рядом с каждой диаграммой или изменением агента, чтобы демонстрации не превосходили возможности системы.
Храните информацию об ответственных лицах, настройках и записях о возврате к предыдущему состоянию рядом с каждой диаграммой или изменением агента, чтобы демонстрации не превосходили возможности системы.
Храните информацию об ответственных лицах, настройках и записях о возврате к предыдущему состоянию рядом с каждой диаграммой или изменением агента, чтобы демонстрации не превосходили возможности системы.
Храните информацию об ответственных лицах, настройках и записях о возврате к предыдущему состоянию рядом с каждой диаграммой или изменением агента, чтобы демонстрации не превосходили возможности системы.
Храните информацию об ответственных лицах, настройках и записях о возврате к предыдущему состоянию рядом с каждой диаграммой или изменением агента, чтобы демонстрации не превосходили возможности системы.
Храните информацию об ответственных лицах, настройках и записях о возврате к предыдущему состоянию рядом с каждой диаграммой или изменением агента, чтобы демонстрации не превосходили возможности системы.
Храните информацию об ответственных лицах, настройках и записях о возврате к предыдущему состоянию рядом с каждой диаграммой или изменением агента, чтобы демонстрации не превосходили возможности системы.
Храните информацию об ответственных лицах, настройках и записях о возврате к предыдущему состоянию рядом с каждой диаграммой или изменением агента, чтобы демонстрации не превосходили возможности системы.
Храните информацию об ответственных лицах, настройках и записях о возврате к предыдущему состоянию рядом с каждой диаграммой или изменением агента, чтобы демонстрации не превосходили возможности системы.
Храните информацию об ответственных лицах, настройках и записях о возврате к предыдущему состоянию рядом с каждой диаграммой или изменением агента, чтобы демонстрации не превосходили возможности системы.
Храните информацию об ответственных лицах, настройках и записях о возврате к предыдущему состоянию рядом с каждой диаграммой или изменением агента, чтобы демонстрации не превосходили возможности системы.
Храните информацию об ответственных лицах, настройках и записях о возврате к предыдущему состоянию рядом с каждой диаграммой или изменением агента, чтобы демонстрации не превосходили возможности системы.
Храните информацию об ответственных лицах, настройках и записях о возврате к предыдущему состоянию рядом с каждой диаграммой или изменением агента, чтобы демонстрации не превосходили возможности системы.
Храните информацию об ответственных лицах, настройках и записях о возврате к предыдущему состоянию рядом с каждой диаграммой или изменением агента, чтобы демонстрации не превосходили возможности системы.
Храните информацию об ответственных лицах, настройках и записях о возврате к предыдущему состоянию рядом с каждой диаграммой или изменением агента, чтобы демонстрации не превосходили возможности системы.
Храните информацию об ответственных лицах, настройках и записях о возврате к предыдущему состоянию рядом с каждой диаграммой или изменением агента, чтобы демонстрации не превосходили возможности системы.
Храните информацию об ответственных лицах, настройках и записях о возврате к предыдущему состоянию рядом с каждой диаграммой или изменением агента, чтобы демонстрации не превосходили возможности системы.
Храните информацию об ответственных лицах, настройках и записях о возврате к предыдущему состоянию рядом с каждой диаграммой или изменением агента, чтобы демонстрации не превосходили возможности системы.
Храните информацию об ответственных лицах, настройках и записях о возврате к предыдущему состоянию рядом с каждой диаграммой или изменением агента, чтобы демонстрации не превосходили возможности системы.
Храните информацию об ответственных лицах, настройках и записях о возврате к предыдущему состоянию рядом с каждой диаграммой или изменением агента, чтобы демонстрации не превосходили возможности системы.
Храните информацию об ответственных лицах, настройках и записях о возврате к предыдущему состоянию рядом с каждой диаграммой или изменением агента, чтобы демонстрации не превосходили возможности системы.
Храните информацию об ответственных лицах, настройках и записях о возврате к предыдущему состоянию рядом с каждой диаграммой или изменением агента, чтобы демонстрации не превосходили возможности системы.
Храните информацию об ответственных лицах, настройках и записях о возврате к предыдущему состоянию рядом с каждой диаграммой или изменением агента, чтобы демонстрации не превосходили возможности системы.
Храните информацию об ответственных лицах, настройках и записях о возврате к предыдущему состоянию рядом с каждой диаграммой или изменением агента, чтобы демонстрации не превосходили возможности системы.
Храните информацию об ответственных лицах, настройках и записях о возврате к предыдущему состоянию рядом с каждой диаграммой или изменением агента, чтобы демонстрации не превосходили возможности системы.
Храните информацию об ответственных лицах, настройках и записях о возврате к предыдущему состоянию рядом с каждой диаграммой или изменением агента, чтобы демонстрации не превосходили возможности системы.
Храните информацию об ответственных лицах, настройках и записях о возврате к предыдущему состоянию рядом с каждой диаграммой или изменением агента, чтобы демонстрации не превосходили возможности системы.
Храните информацию об ответственных лицах, настройках и записях о возврате к предыдущему состоянию рядом с каждой диаграммой или изменением агента, чтобы демонстрации не превосходили возможности системы.
Храните информацию об ответственных лицах, настройках и записях о возврате к предыдущему состоянию рядом с каждой диаграммой или изменением агента, чтобы демонстрации не превосходили возможности системы.
Храните информацию об ответственных лицах, настройках и записях о возврате к предыдущему состоянию рядом с каждой диаграммой или изменением агента, чтобы демонстрации не превосходили возможности системы.
Храните информацию об ответственных лицах, настройках и записях о возврате к предыдущему состоянию рядом с каждой диаграммой или изменением агента, чтобы демонстрации не превосходили возможности системы.
Храните информацию об ответственных лицах, настройках и записях о возврате к предыдущему состоянию рядом с каждой диаграммой или изменением агента, чтобы демонстрации не превосходили возможности системы.
Храните информацию об ответственных лицах, настройках и записях о возврате к предыдущему состоянию рядом с каждой диаграммой или изменением агента, чтобы демонстрации не превосходили возможности системы.
Храните информацию об ответственных лицах, настройках и записях о возврате к предыдущему состоянию рядом с каждой диаграммой или изменением агента, чтобы демонстрации не превосходили возможности системы.
Храните информацию об ответственных лицах, настройках и записях о возврате к предыдущему состоянию рядом с каждой диаграммой или изменением агента, чтобы демонстрации не превосходили возможности системы.
Храните информацию об ответственных лицах, настройках и записях о возврате к предыдущему состоянию рядом с каждой диаграммой или изменением агента, чтобы демонстрации не превосходили возможности системы.
Храните информацию об ответственных лицах, настройках и примечаниях по возврату к предыдущему состоянию рядом с каждой диаграммой или изменением агента, чтобы демонстрации не превосходили возможности системы.
Храните информацию об ответственных лицах, настройках и примечаниях по возврату к предыдущему состоянию рядом с каждой диаграммой или изменением агента, чтобы демонстрации не превосходили возможности системы.
Храните информацию об ответственных лицах, настройках и примечаниях по возврату к предыдущему состоянию рядом с каждой диаграммой или изменением агента, чтобы демонстрации не превосходили возможности системы.
Храните информацию об ответственных лицах, настройках и примечаниях по возврату к предыдущему состоянию рядом с каждой диаграммой или изменением агента, чтобы демонстрации не превосходили возможности системы.
Храните информацию об ответственных лицах, настройках и примечаниях по возврату к предыдущему состоянию рядом с каждой диаграммой или изменением агента, чтобы демонстрации не превосходили возможности системы.
Храните информацию об ответственных лицах, настройках и примечаниях по возврату к предыдущему состоянию рядом с каждой диаграммой или изменением агента, чтобы демонстрации не превосходили возможности системы.
Храните информацию об ответственных лицах, настройках и примечаниях по возврату к предыдущему состоянию рядом с каждой диаграммой или изменением агента, чтобы демонстрации не превосходили возможности системы.