Публикация решения: update main.py

main.py

@@ -1,96 +1,49 @@
-"""Console entry point that prints LangGraph agent output in stream mode."""
+"""Console demo for stream-mode LangGraph agent."""
 
-from __future__ import annotations
+import os
-
-import sys
 from typing import Any
 
+from langchain_core.messages import HumanMessage, AIMessage
+from langgraph.graph import StateGraph
+
 from agent import agent
 
-
+# Helper to format messages
-_current_step: int | None = None
-
 
 def format_message(message: Any) -> str:
-    """Format a completed model message or a tool-call request."""
+    if hasattr(message, "content") and message.content:
-
+        return message.content
-    content = getattr(message, "content", "")
+    # fallback for tool calls
-    if content:
+    if hasattr(message, "tool_calls") and message.tool_calls:
-        return str(content)
+        call = message.tool_calls[0]
-
-    tool_calls = getattr(message, "tool_calls", None) or []
-    if tool_calls:
-        call = tool_calls[0]
         return f"{call['name']}({call['args']})"
-
     return ""
 
+# Stream the conversation
 
-def format_chunk_message(chunk: tuple[Any, dict[str, Any]]) -> None:
+def main() -> None:
-    """Print token chunks and separate LangGraph steps."""
+    # initial human message
-
+    human_msg = HumanMessage(content="Составь список покупок: молоко, хлеб, яблоки")
-    global _current_step
+    # start stream
-
-    message, meta = chunk
-    langgraph_step = meta.get("langgraph_step")
-    if _current_step is None:
-        _current_step = langgraph_step
-    elif langgraph_step != _current_step:
-        _current_step = langgraph_step
-        print("\n--- --- ---\n", end="")
-
-    if message.content:
-        print(message.content, end="", flush=True)
-
-
-def run_stream(query: str) -> None:
-    """Run the agent with .stream() and handle messages/updates chunks."""
-
-    global _current_step
-    _current_step = None
-
     stream = agent.stream(
-        {"messages": [{"role": "human", "content": query}]},
+        {"messages": [human_msg]},
         stream_mode=["messages", "updates"],
     )
 
+    step = 1
     for chunk in stream:
         chunk_type, chunk_data = chunk
-
         if chunk_type == "messages":
-            format_chunk_message(chunk_data)
+            message, meta = chunk_data
-
+            if meta.get("langgraph_step") != step:
-        if chunk_type == "updates":
+                step = meta.get("langgraph_step")
-            model_update = chunk_data.get("model")
+                print("\n--- --- ---\n")
-            if model_update:
+            if message.content:
-                last_message = model_update["messages"][-1]
+                print(message.content, end="", flush=True)
-                formatted = format_message(last_message)
+        elif chunk_type == "updates":
-                if formatted:
+            if chunk_data.get("model"):
-                    print(formatted, end="", flush=True)
+                last_msg = chunk_data["model"]["messages"][-1]
-
+                print(format_message(last_msg))
-    print()
-
-
-def main() -> None:
-    """Read one CLI query or start a small interactive loop."""
-
-    if len(sys.argv) > 1:
-        run_stream(" ".join(sys.argv[1:]))
-        return
-
-    print("Stream-mode AI agent. Type 'exit' to quit.")
-    while True:
-        try:
-            query = input("You: ").strip()
-        except EOFError:
-            print()
-            break
-
-        if query.lower() in {"exit", "quit"}:
-            break
-        if query:
-            run_stream(query)
-
 
 if __name__ == "__main__":
     main()