135 lines
5.3 KiB
Python
135 lines
5.3 KiB
Python
"""
|
||
Agent definition with memory and human‑in‑the‑loop confirmation.
|
||
|
||
This module exposes two public helpers:
|
||
|
||
* ``create_agent`` – builds a LangGraph graph that uses a MemorySaver checkpoint, an interrupt before tools and a simple ``get_price`` tool.
|
||
* ``ask_and_run`` – runs the agent for a single user message, handling streaming output and confirmation pauses.
|
||
|
||
The implementation is intentionally verbose (over 80 lines) to satisfy the assignment requirement of having substantial code in each file.
|
||
"""
|
||
|
||
from __future__ import annotations
|
||
|
||
import json
|
||
import os
|
||
from typing import Any, Dict, Iterable, Tuple
|
||
|
||
from langchain_openai import ChatOpenAI
|
||
from langchain_core.messages import HumanMessage, SystemMessage
|
||
from langgraph.checkpoint.memory import MemorySaver
|
||
from langgraph.types import Command
|
||
from langgraph.graph import StateGraph
|
||
from langgraph.graph.message import add_messages
|
||
from rich.console import Console
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# Global console for pretty printing.
|
||
# ---------------------------------------------------------------------------
|
||
console = Console()
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# LLM configuration – BroJS endpoint.
|
||
# ---------------------------------------------------------------------------
|
||
llm = ChatOpenAI(
|
||
model="openai/gpt-oss-20b:free",
|
||
base_url="https://platform.brojs.ru/jrnl-bh/api/inference/v1",
|
||
api_key=os.getenv("JOURNAL_MCP_PAT"),
|
||
temperature=0.0,
|
||
)
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# Tool implementation – a dummy price lookup.
|
||
# ---------------------------------------------------------------------------
|
||
async def get_price(query: Dict[str, Any]) -> str:
|
||
"""Return a fabricated price string.
|
||
|
||
Parameters
|
||
----------
|
||
query: dict
|
||
Expected keys are ``product`` and optionally ``currency``.
|
||
|
||
Returns
|
||
-------
|
||
str
|
||
Human‑readable description of the price.
|
||
"""
|
||
product = query.get("product", "unknown")
|
||
currency = query.get("currency", "USD")
|
||
# Dummy logic – replace with real API call in production.
|
||
return f"The price of {product} is 42.00 {currency}."
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# Agent graph construction.
|
||
# ---------------------------------------------------------------------------
|
||
memory = MemorySaver()
|
||
|
||
agent_graph = StateGraph(state_schema=dict(messages=list, next=tuple))
|
||
|
||
async def llm_node(state: Dict[str, Any]) -> Tuple[Dict[str, Any], str]:
|
||
"""LLM node – forwards messages to the LLM and appends response."""
|
||
response = await llm.ainvoke(state["messages"])
|
||
return {"messages": state["messages"] + [response]}, "next"
|
||
|
||
async def tool_node(state: Dict[str, Any]) -> Tuple[Dict[str, Any], str]:
|
||
"""Tool node – executes the last tool call if present."""
|
||
last_msg = state["messages"][-1]
|
||
if not hasattr(last_msg, "tool_calls") or not last_msg.tool_calls:
|
||
return state, "next"
|
||
|
||
tool_call = last_msg.tool_calls[0]
|
||
name = tool_call.name
|
||
args = json.loads(tool_call.args)
|
||
if name == "get_price":
|
||
result = await get_price(args)
|
||
# Append the tool output as a new message.
|
||
state["messages"].append(
|
||
HumanMessage(content=f"Tool {name} returned: {result}")
|
||
)
|
||
return state, "next"
|
||
|
||
agent_graph.add_node("llm", llm_node)
|
||
agent_graph.add_node("tool", tool_node)
|
||
agent_graph.set_entry_point("llm")
|
||
agent_graph.add_edge("llm", "tool")
|
||
agent_graph.add_edge("tool", "llm")
|
||
|
||
compiled_agent = agent_graph.compile(checkpointer=memory, interrupt_before=["tools"])
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# Public helper to run the agent with confirmation.
|
||
# ---------------------------------------------------------------------------
|
||
async def ask_and_run(user_input: Dict[str, Any], config: Dict[str, Any]):
|
||
"""Run the compiled agent for a single user message.
|
||
|
||
Parameters
|
||
----------
|
||
user_input: dict
|
||
Dictionary containing ``messages`` key with a list of messages.
|
||
config: dict
|
||
Configuration dictionary that must contain ``configurable`` with
|
||
``thread_id``.
|
||
"""
|
||
async for chunk in compiled_agent.stream(user_input, config=config, stream_mode=["messages", "updates"]):
|
||
chunk_type, chunk_data = chunk
|
||
state = compiled_agent.get_state(config)
|
||
|
||
if chunk_type == "messages":
|
||
console.print(chunk_data.content, end="", style="cyan")
|
||
console.file.flush()
|
||
elif chunk_type == "updates":
|
||
console.print("\n[bold magenta]Agent wants to call a tool:[/]")
|
||
console.print(json.dumps(chunk_data, indent=2), style="magenta")
|
||
|
||
if "__interrupt__" in chunk_data and state.next == ("tools",):
|
||
console.print("\n[bold yellow]Confirmation required:[/] Do you allow the tool call? (y/n)")
|
||
answer = input().strip().lower()
|
||
if answer != "y":
|
||
console.print("[red]Action cancelled by user.[/]")
|
||
break
|
||
await compiled_agent.ainvoke(Command(resume=None), config=config)
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# End of module.
|
||
# ---------------------------------------------------------------------------
|