fluzzi32/connpy
1
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

📌 1. Uncommitted Changes (Staged for next commit)

Browse Source 
Focus: UI stability and efficient rendering.
   * Markdown Rendering Rewrite: Removed the dependency on rich.live.Live (which caused flickering and
     high CPU usage by constantly re-rendering the entire panel).
   * New BlockMarkdownRenderer: Implemented in printer.py (alias IncrementalMarkdownParser), it
     accumulates text in a buffer and only prints to the screen when it detects a complete block (e.g.,
     line breaks or
  ` code blocks).
   * UI Optimizations (terminal_ui.py & stubs.py): Waiting spinners now stop cleanly, and the UI
     transitions smoothly to block printing. Fixed visual truncation issues in the bottom "Tab prompt"
     bar for excessively long commands.

  📦 2. Commit History (Last 4 Commits)

  9446baf - improve ai rules
   * Strict Anti-Hallucination (ai.py): Injected MANDATORY rules into the System Prompt for both
     Architect and Engineer agents. Now, if the terminal buffer is empty or only contains an idle prompt
     (e.g., iol#), the AI is strictly forced to state that it lacks data instead of inventing topologies
     or configurations.
   * Language Preference: Explicitly instructed agents to always respond in the same language the user
     used to ask the question.

  64377f7 - move context block logic to server and improvements
   * Context Precision (ai_service.py): Moved context partitioning logic to the service/server side. It
     now calculates exact start_pos and end_pos based on identified commands, preventing mixed outputs
     or residual text from bleeding into the AI's prompt.
   * Token Savings (server.py): The server now selectively strips garbage metadata and UI caches that
     add no value to the LLM before sending the payload over the wire.

  e4fd1ad - fix logclean for 6wind
   * Full ANSI/CSI Support (utils.py): Replaced the legacy rigid escape filter with a complete CSI
     (Control Sequence Introducer) parser. The client can now accurately process numeric cursor
     movements (C, D), inline dynamic erasures (K), and absolute shifts (G), ensuring connpy understands
     exactly what a 6WIND router or other VNFs render on the screen without garbage characters.

  b0a914a - updates al copilot
   * Copilot <> gRPC Integration: Implemented the async plumbing required for the Copilot to work over
     the gRPC tunnel (server.py & stubs.py).
   * Initial Streaming UI: Added the first iteration of chunk callbacks to provide real-time feedback
     while the LLM generates responses, laying the groundwork for the uncommitted block-renderer
     optimizations.
This commit is contained in:
Fede Luzzi
2026-05-18 14:07:24 -03:00
parent 9446bafc0c
commit 5a8b744aa8
6 changed files with 210 additions and 94 deletions
Download Patch File Download Diff File Expand all files Collapse all files
connpy/_version.py
+1 -1
View File
@@ -1 +1 @@
__version__ = "6.0.0b8"
__version__ = "6.0.0b9"
connpy/ai.py
+15 -24
View File
@@ -299,7 +299,6 @@ class ai:
- response: reconstructed ModelResponse (same as non-streaming)
- streamed: True if text was rendered to console during streaming
"""
from rich.live import Live
stream_resp = completion(model=model, messages=messages, tools=tools, api_key=api_key, stream=True, **kwargs)
@@ -307,7 +306,7 @@ class ai:
full_content = ""
is_streaming_text = False
has_tool_calls = False
live_display = None
header_printed = False
# Determine styling based on current brain
role_label = "Network Architect" if "architect" in label.lower() else "Network Engineer"
@@ -336,7 +335,6 @@ class ai:
if not chunk_callback:
if not is_streaming_text:
# Stop spinner definitively
if status:
try:
status.stop()
@@ -345,35 +343,28 @@ class ai:
# Create a stable, direct Console to bypass _ConsoleProxy recreation bugs
from rich.console import Console as RichConsole
from .printer import connpy_theme, get_original_stdout
from rich.rule import Rule
from .printer import connpy_theme, get_original_stdout, IncrementalMarkdownParser
stable_console = RichConsole(theme=connpy_theme, file=get_original_stdout())
live_display = Live(
Panel(Markdown(full_content), title=title, border_style=border, expand=False),
console=stable_console,
refresh_per_second=8,
transient=False
)
live_display.start()
stable_console.print(Rule(f"[bold {border}]{title}[/bold {border}]", style=border))
header_printed = True
md_parser = IncrementalMarkdownParser(console=stable_console)
is_streaming_text = True
else:
live_display.update(
Panel(Markdown(full_content), title=title, border_style=border, expand=False)
)
md_parser.feed(delta.content)
except Exception as e:
if not chunks:
raise
finally:
if live_display:
# Render final state with complete content
if header_printed:
try:
live_display.update(
Panel(Markdown(full_content), title=title, border_style=border, expand=False)
)
except Exception:
pass
try:
live_display.stop()
md_parser.flush()
from rich.console import Console as RichConsole
from rich.rule import Rule
from .printer import connpy_theme, get_original_stdout
stable_console = RichConsole(theme=connpy_theme, file=get_original_stdout())
stable_console.print(Rule(style=border))
except Exception:
pass
connpy/cli/terminal_ui.py
+83 -25
View File
@@ -12,7 +12,6 @@ from textwrap import dedent
from rich.console import Console
from rich.panel import Panel
from rich.markdown import Markdown
from rich.live import Live
from prompt_toolkit import PromptSession
from prompt_toolkit.key_binding import KeyBindings
from prompt_toolkit.formatted_text import HTML
@@ -174,7 +173,40 @@ class CopilotInterface:
base_str = f'\u25b6 Ctrl+\u2191/\u2193 adjusts by 50 lines [Tab: {m_label}]'
else:
idx = max(0, state['total_cmds'] - state['context_cmd'])
desc = blocks[idx][2]
import re
def clean_preview(text):
# Limpia saltos de línea y el prompt inicial (todo hasta #, > o $) para que quede solo el comando
original = text.strip().replace('\r', '').replace('\n', ' ')
cleaned = re.sub(r'^.*?[#>\$]\s*', '', original)
# Si limpiar el prompt nos deja con un string vacío (ej: era solo "iol#"), devolvemos el original
return cleaned if cleaned else original
if state['context_mode'] == self.mode_range:
range_blocks = blocks[idx:]
# Si hay más de un bloque, el último es siempre el prompt vacío/actual. Lo omitimos visualmente.
if len(range_blocks) > 1:
range_blocks = range_blocks[:-1]
# Limpiar y truncar comandos muy largos para que no rompan la UI
previews = []
for b in range_blocks:
p = clean_preview(b[2])
if p:
# Truncar comandos individuales largos
if len(p) > 25: p = p[:22] + "..."
previews.append(p)
if not previews:
desc = clean_preview(blocks[idx][2])
elif len(previews) <= 3:
desc = " + ".join(previews)
else:
desc = f"{previews[0]} + {previews[1]} + {previews[2]} ... (+{len(previews)-3})"
else:
# Modo SINGLE original
desc = clean_preview(blocks[idx][2])
base_str = f'\u25b6 {desc} [Tab: {m_label}]'
# Wrap base_str in a style to maintain consistency and avoid glitches
@@ -304,36 +336,63 @@ class CopilotInterface:
persona_title = "Network Architect" if active_persona == "architect" else "Network Engineer"
active_buffer = get_active_buffer()
live_text = "Thinking..."
panel = Panel(live_text, title=f"[bold {persona_color}]{persona_title}[/bold {persona_color}]", border_style=persona_color)
live_text = ""
first_chunk = True
import sys
from rich.rule import Rule
from rich.status import Status
from connpy.printer import IncrementalMarkdownParser
md_parser = IncrementalMarkdownParser(console=self.console)
status_spinner = Status(
f"[bold {persona_color}]{persona_title}:[/bold {persona_color}] [dim]Thinking...[/dim]",
console=self.console,
spinner="dots"
)
status_spinner.start()
def on_chunk(text):
nonlocal live_text
if live_text == "Thinking...": live_text = ""
nonlocal live_text, first_chunk
if first_chunk:
status_spinner.stop()
# Print header rule before first chunk arrives
self.console.print(Rule(
f"[bold {persona_color}]{persona_title}[/bold {persona_color}]",
style=persona_color
))
first_chunk = False
live_text += text
md_parser.feed(text)
with Live(panel, console=self.console, refresh_per_second=10) as live:
def update_live(t):
live.update(Panel(Markdown(t), title=f"[bold {persona_color}]{persona_title}[/bold {persona_color}]", border_style=persona_color))
wrapped_chunk = lambda t: (on_chunk(t), update_live(live_text))
# Check for interruption during AI call
ai_task = asyncio.create_task(on_ai_call(active_buffer, clean_question, wrapped_chunk, merged_node_info))
try:
while not ai_task.done():
await asyncio.sleep(0.05)
result = await ai_task
except asyncio.CancelledError:
return "cancel", None, None
# Check for interruption during AI call
ai_task = asyncio.create_task(on_ai_call(active_buffer, clean_question, on_chunk, merged_node_info))
try:
while not ai_task.done():
await asyncio.sleep(0.05)
result = await ai_task
except asyncio.CancelledError:
status_spinner.stop()
return "cancel", None, None
# Ensure spinner is stopped if no chunks arrived
if first_chunk:
status_spinner.stop()
# Close the streamed output with a Rule
if not first_chunk:
md_parser.flush()
self.console.print(Rule(style=persona_color))
if not result or result.get("error"):
if result and result.get("error"): self.console.print(f"[red]Error: {result['error']}[/red]")
if first_chunk and result and result.get("error"):
self.console.print(f"[red]Error: {result['error']}[/red]")
return "cancel", None, None
# 4. Handle result
if live_text == "Thinking..." and result.get("guide"):
# If no chunks were streamed but we have a guide, print it as a panel
if first_chunk and result and result.get("guide"):
self.console.print(Panel(Markdown(result["guide"]), title=f"[bold {persona_color}]{persona_title}[/bold {persona_color}]", border_style=persona_color))
commands = result.get("commands", [])
@@ -437,5 +496,4 @@ class CopilotInterface:
finally:
state['cancelled'] = True
self.console.print("[dim]Returning to session...[/dim]")
connpy/core.py
+1
View File
@@ -689,6 +689,7 @@ class node:
continue
break
finally:
print("\033[2m Returning to session...\033[0m", flush=True)
# Reiniciar el lector de la terminal para volver al modo interactivo SSH/Telnet
if hasattr(stream, 'start_reading'):
stream.start_reading()
connpy/grpc_layer/stubs.py
+28 -43
View File
@@ -102,6 +102,7 @@ class NodeStub:
with copilot_terminal_mode():
action, commands, custom_cmd = asyncio.run(run_remote_copilot())
print("\033[2m Returning to session...\033[0m", flush=True)
# Prepare final action for server
action_sent = "cancel"
if action == "send_all" and commands:
@@ -726,7 +727,6 @@ class AIStub:
import queue
from rich.prompt import Prompt
from rich.text import Text
from rich.live import Live
from rich.panel import Panel
from rich.markdown import Markdown
@@ -757,7 +757,7 @@ class AIStub:
responses = self.stub.ask(request_generator())
full_content = ""
live_display = None
header_printed = False
final_result = {"response": "", "chat_history": []}
# Background thread to pull responses from gRPC into a local queue
@@ -822,69 +822,53 @@ class AIStub:
if response.debug_message:
if debug:
if live_display:
try: live_display.stop()
except: pass
if status:
try: status.stop()
except: pass
printer.console.print(Text.from_ansi(response.debug_message))
if live_display:
try: live_display.start()
except: pass
elif status:
if status:
try: status.start()
except: pass
continue
if response.important_message:
if live_display:
try: live_display.stop()
except: pass
if status:
try: status.stop()
except: pass
printer.console.print(Text.from_ansi(response.important_message))
if live_display:
try: live_display.start()
except: pass
elif status:
if status:
try: status.start()
except: pass
continue
if not response.is_final:
if response.text_chunk:
full_content += response.text_chunk
if not live_display:
if not header_printed:
if status:
try: status.stop()
except: pass
from rich.console import Console as RichConsole
from ..printer import connpy_theme, get_original_stdout
from rich.rule import Rule
from ..printer import connpy_theme, get_original_stdout, IncrementalMarkdownParser
stable_console = RichConsole(theme=connpy_theme, file=get_original_stdout())
# We default to Engineer title during stream, final result will correct it if needed
live_display = Live(
Panel(Markdown(full_content), title="[bold engineer]Network Engineer[/bold engineer]", border_style="engineer", expand=False),
console=stable_console,
refresh_per_second=8,
transient=False
)
live_display.start()
else:
live_display.update(
Panel(Markdown(full_content), title="[bold engineer]Network Engineer[/bold engineer]", border_style="engineer", expand=False)
)
# Print header on first chunk
stable_console.print(Rule("[bold engineer]Network Engineer[/bold engineer]", style="engineer"))
header_printed = True
# Initialize parser
md_parser = IncrementalMarkdownParser(console=stable_console)
full_content += response.text_chunk
md_parser.feed(response.text_chunk)
continue
if response.is_final:
if live_display:
try: live_display.stop()
except: pass
# Final stop for status to ensure it disappears before the panel
if header_printed:
from rich.rule import Rule
md_parser.flush()
if status:
try: status.stop()
except: pass
@@ -895,13 +879,14 @@ class AIStub:
role_label = "Network Architect" if responder == "architect" else "Network Engineer"
title = f"[bold {alias}]{role_label}[/bold {alias}]"
content_to_print = full_content or final_result.get("response", "")
if content_to_print:
if live_display:
# Re-render the final frame with correct title/colors
live_display.update(Panel(Markdown(content_to_print), title=title, border_style=alias, expand=False))
else:
printer.console.print(Panel(Markdown(content_to_print), title=title, border_style=alias, expand=False))
if header_printed:
from rich.console import Console as RichConsole
from ..printer import connpy_theme, get_original_stdout
stable_console = RichConsole(theme=connpy_theme, file=get_original_stdout())
stable_console.print(Rule(style=alias))
elif not full_content and final_result.get("response"):
# If nothing streamed but we have response (e.g. error or direct guide)
printer.console.print(Panel(Markdown(final_result["response"]), title=title, border_style=alias, expand=False))
break
except Exception as e:
# Check if it was a gRPC error that we should let handle_errors catch
connpy/printer.py
+82 -1
View File
@@ -15,7 +15,17 @@ class ThreadLocalStream:
def write(self, data):
stream = self._get_stream()
if stream:
stream.write(data)
import time
retries = 0
while True:
try:
stream.write(data)
break
except BlockingIOError:
if retries > 50:
raise
time.sleep(0.01)
retries += 1
def flush(self):
stream = self._get_stream()
@@ -496,3 +506,74 @@ class _ThemeProxy:
return getattr(local.theme, name)
connpy_theme = _ThemeProxy()
class BlockMarkdownRenderer:
"""
Block-buffered streaming markdown renderer.
Accumulates text until block boundaries are detected,
then renders complete blocks using Rich's Markdown.
"""
def __init__(self, console=None):
from rich.console import Console as RichConsole
from .printer import connpy_theme, get_original_stdout
self._console = console or RichConsole(
theme=connpy_theme, file=get_original_stdout()
)
self._line_buf = "" # chars waiting for \n
self._block_lines = [] # complete lines for current block
self._in_code_block = False
def feed(self, text):
self._line_buf += text
while '\n' in self._line_buf:
idx = self._line_buf.index('\n')
line = self._line_buf[:idx + 1]
self._line_buf = self._line_buf[idx + 1:]
self._process_line(line)
def flush(self):
if self._line_buf:
self._block_lines.append(self._line_buf)
self._line_buf = ""
self._flush_block()
def _process_line(self, line):
stripped = line.strip()
if stripped.startswith('```'):
if not self._in_code_block:
# Flush accumulated text before code block
self._flush_block()
self._in_code_block = True
self._block_lines.append(line)
else:
# Include closing fence and flush code block
self._block_lines.append(line)
self._in_code_block = False
self._flush_block()
return
if self._in_code_block:
self._block_lines.append(line)
return
# Blank line = paragraph break
if stripped == '':
self._block_lines.append(line)
self._flush_block()
return
self._block_lines.append(line)
def _flush_block(self):
if not self._block_lines:
return
block_text = ''.join(self._block_lines).strip()
self._block_lines = []
if not block_text:
return
from rich.markdown import Markdown
self._console.print(Markdown(block_text))
# Alias for backward compatibility
IncrementalMarkdownParser = BlockMarkdownRenderer