PlantUML 在 LLM Agent 中的角色：让 AI 推理过程可视化与人机协同

import langchain
from langchain.agents import AgentExecutor
from langchain.tools import tool

langchain.debug = True

agent = AgentExecutor(...)
result = agent.invoke({"input": "What is PlantUML?"})

from langchain.callbacks.base import BaseCallbackHandler

class PlantUMLCallback(BaseCallbackHandler):
    def __init__(self):
        self.steps = []
    
    def on_agent_action(self, action, **kwargs):
        self.steps.append(f"Action: {action.tool}, Input: {action.tool_input}")
    
    def on_agent_finish(self, finish, **kwargs):
        self.steps.append(f"Finish: {finish.return_values}")
    
    def to_puml(self):
        puml = "@startuml\nstart\n"
        for s in self.steps:
            puml += f":{s};\n"
        puml += "stop\n@enduml\n"
        return puml

# 用法
cb = PlantUMLCallback()
agent.invoke({"input": "..."}, config={"callbacks": [cb]})
print(cb.to_puml())
# 渲染 → 在浏览器看决策过程

import plantuml
puml_server = plantuml.PlantUML(url='http://www.plantuml.com/plantuml/svg/')

def render_trace(steps):
    puml = "@startuml\nstart\n"
    for s in steps:
        puml += f":{s};\n"
    puml += "stop\n@enduml\n"
    return puml_server.processes(puml)

# 调 /var/log/agent-trace.svg，方便 devops 看

# LangGraph
from langgraph.graph import StateGraph

graph = StateGraph(MyState)
graph.add_node("search", search_node)
graph.add_node("rank", rank_node)
graph.add_edge("search", "rank")

# 手动导出 graph.adjacency 为 PlantUML
def to_puml(graph):
    puml = "@startuml\n"
    for src, dst in graph.edges:
        puml += f"{src} --> {dst}\n"
    puml += "@enduml"
    return puml

with open("/tmp/agent-graph.puml", "w") as f:
    f.write(to_puml(graph))

crew = Crew(agents=[...], tasks=[...])
crew.plot()  # 输出 tasks_graph.png

def crew_to_puml(crew):
    puml = "@startuml\n"
    for agent in crew.agents:
        puml += f'rectangle "{agent.role}" as {agent.id}\n'
    for task in crew.tasks:
        for agent in task.assigned_agents:
            puml += f"{agent.id} --> {task.id}\n"
    puml += "@enduml"
    return puml

class MultiAgentSystem:
    def __init__(self):
        self.transitions = []
    
    def transition(self, from_agent, to_agent, reason):
        self.transitions.append({
            "from": from_agent,
            "to": to_agent,
            "reason": reason
        })
    
    def render(self):
        puml = "@startuml\n"
        # 加所有 agent
        for agent in set(a["from"] for a in self.transitions) | set(a["to"] for a in self.transitions):
            puml += f'rectangle "{agent}" as {agent}\n'
        for t in self.transitions:
            puml += f"{t['from']} --> {t['to']} : {t['reason']}\n"
        puml += "@enduml"
        return puml

mas = MultiAgentSystem()
mas.transition("user", "router", "ask plan")
mas.transition("router", "researcher", "needs data")
mas.transition("researcher", "critic", "review")
# ...
print(mas.render())

class InterventionHook:
    def __init__(self, mas):
        self.mas = mas
    
    def check(self, agent_output):
        for rule in self.rules:
            if rule.matches(agent_output):
                return HumanInterventionRequest(
                    agent=agent_output.agent,
                    reason=rule.reason,
                    diagram=self.mas.render()  # 当前状态
                )

# debug_agent.py
from langchain_community.tools import DuckDuckGoSearchRun
from my_callback import PlantUMLCallback

cb = PlantUMLCallback()
tools = [DuckDuckGoSearchRun()]

agent = initialize_agent(
    tools,
    llm,
    agent=AgentType.ZERO_SHOT_REACT_DESCRIPTION,
    verbose=True,
    callbacks=[cb]
)

try:
    result = agent.run("Tell me about plantuml.com")
except Exception as e:
    print("Error:", e)
finally:
    print(cb.to_puml())
    # → 写到 /var/log/agent-trace.svg