CopilotKit 源码分析

发表于 2026-08-04

CopilotKit 整体架构总览

前端组件如何注册到后端agent服务中的

通过CopilotKit 生成式UI框架了解到，根据开放程度可以分成3类组件

controrolled Generative UI
declarative Generative UI
open Generative UI

其中第2和3类都是在CopilotRuntime 中注册的，注册后在后续发给agent的http请求中会以上下文的方式(其实也是参数)传递给agent

onst langGraphAgent = new LangGraphHttpAgent({ url: "http://localhost:8004" });

const runtime = new CopilotRuntime({
  agents: { default: langGraphAgent },
  a2ui: { injectA2UITool: true }, // 启用 declarative Generative UI
  openGenerativeUI: true, // 启用 open Generative UI  代理可以生成任意类型的用户界面——包括 HTML、CSS、JavaScript 等代码，并可以直接在聊天界面中展示
  mcpApps: { // 注册 MCP 应用程序
    servers: [
      {
        type: "http",
        url: "https://mcp.excalidraw.com", // <- Exalidraw MCP Server
        serverId: "example_mcp_server",
      },
    ],
  },
});

const app = createCopilotEndpoint({
  runtime,
  basePath: "/api/copilotkit",
});

serve({ fetch: app.fetch, port: 4004 }, () => {
  console.log("\u2713 CopilotKit API server running at http://localhost:4004");
});

三类组件注册流程总览

源码分析：
已知CopilotKit 是一个基于agui协议的框架,agui协议是基于http的上层协议
LangGraphHttpAgent 从httpAgent 中继承
本身的作用就是通过http协议与agent进行通信

The HttpAgent extends AbstractAgent to provide HTTP-based connectivity to remote AI agents. It handles the request/response cycle and transforms the HTTP event stream into standard Agent User Interaction Protocol events.

CopilotRuntime 是一个桥接器，用于前端和后端agent服务之间的通信，控制请求发送，信息处理等操作

createCopilotEndpoint 会创建一个node服务，利用CopilotRuntime实例，将tools 信息给到agent 并处理请求过程中的信息，返回给前端

对于第1类使用 useComponent 进行组件注册的过程会使用useFrontendTool 方法，将组件注册到copilotkit 实例的tools 中

对于使用组件传入的tools 在实例化过程中会自动注册到copilotkit 实例的tools 中

从输入框提交信息后会经过copiloykit 实例发起runAgent

在初始化创建的时候就会传入tools, 后续在connectAgent 和 runAgent 中都会将tools 信息传递给后端agent

agent 本身是一个http-agent

var ProxiedCopilotRuntimeAgent = class ProxiedCopilotRuntimeAgent extends HttpAgent {
  constructor(options) {
    super(options);
  }
}

多样的UI界面是谁渲染的

UI 渲染总览流程

UI 的渲染依然通过前端组件渲染，只不过在渲染时会根据messages 中的返回信息决定使用组件的类型，选择不同的组件进行渲染。
因此, agent 会决定渲染的组件类型，前端组件只是根据组件类型进行渲染。

源码分析：

 const BoundMessageView = renderSlot(messageView, CopilotChatMessageView, {
    messages,
    isRunning,
  });
  ...
    if (children) {
    return (
      <div data-copilotkit style={{ display: "contents" }}>
        {children({
          messageView: BoundMessageView,
          input: BoundInput,
          scrollView: BoundScrollView,
          suggestionView: BoundSuggestionView ?? <></>,
        })}
      </div>
    );
  }

------------------------------CopilotChatMessageView---------------------------------------------

 // Build the flat element list only when we're not virtualizing (avoids
  // creating 500 React elements that we'd immediately discard).
  const messageElements: React.ReactElement[] = shouldVirtualize
    ? []
    : deduplicatedMessages.flatMap(renderMessageBlock);

  // ---------------------------------------------------------------------------
  // children render prop (custom layout, always non-virtual)
  // ---------------------------------------------------------------------------
  if (children) {
    return (
      <div data-copilotkit style={{ display: "contents" }}>
        {children({ messageElements, messages, isRunning, interruptElement })}
      </div>
    );
  }

renderMessageBlock 是一个函数，用于根据消息的类型渲染不同的组件

controlled Generative UI 渲染逻辑

第一类渲染时走 role: “assistant”

const { Component: AssistantComponent, slotProps: assistantSlotProps } =
useMemo(
    () => resolveSlotComponent(assistantMessage, CopilotChatAssistantMessage),
    [assistantMessage],
);

-----------------------------CopilotChatAssistantMessage--------------------------

const boundToolCallsView = renderSlot(
    toolCallsView,
    CopilotChatToolCallsView,
    {
      message,
      messages,
    },
  );

export function CopilotChatToolCallsView({
  message,
  messages = [],
}: CopilotChatToolCallsViewProps) {
  const renderToolCall = useRenderToolCall();

  if (!message.toolCalls || message.toolCalls.length === 0) {
    return null;
  }

  return (
    <>
      {message.toolCalls.map((toolCall) => {
        const toolMessage = messages.find(
          (m) => m.role === "tool" && m.toolCallId === toolCall.id,
        ) as ToolMessage | undefined;

        return (
          <React.Fragment key={toolCall.id}>
            {renderToolCall({
              toolCall,
              toolMessage,
            })}
          </React.Fragment>
        );
      })}
    </>
  );
}

const renderToolCall = useCallback(
    ({
      toolCall,
      toolMessage,
    }: UseRenderToolCallProps): React.ReactElement | null => {
      // Find the render config for this tool call by name
      // For rendering, we show all tool calls regardless of agentId
      // The agentId scoping only affects handler execution (in core)
      // Priority order:
      // 1. Exact match by name (prefer agent-specific if multiple exist)
      // 2. Wildcard (*) renderer
      const exactMatches = renderToolCalls.filter(
        (rc) => rc.name === toolCall.function.name,
      );

      // If multiple renderers with same name exist, prefer the one matching our agentId
      const renderConfig =
        exactMatches.find((rc) => rc.agentId === agentId) ||
        exactMatches.find((rc) => !rc.agentId) ||
        exactMatches[0] ||
        renderToolCalls.find((rc) => rc.name === "*");

      if (!renderConfig) {
        return null;
      }

      const RenderComponent = renderConfig.render;
      const isExecuting = executingToolCallIds.has(toolCall.id);

      // Use the memoized ToolCallRenderer component to prevent unnecessary re-renders
      return (
        <ToolCallRenderer
          key={toolCall.id}
          toolCall={toolCall}
          toolMessage={toolMessage}
          RenderComponent={RenderComponent}
          isExecuting={isExecuting}
        />
      );
    },
    [renderToolCalls, executingToolCallIds, agentId],
  );

const ToolCallRenderer = React.memo(
  function ToolCallRenderer({
    toolCall,
    toolMessage,
    RenderComponent,
    isExecuting,
  }: ToolCallRendererProps) {
    // Memoize args based on the arguments string to maintain stable reference
    const args = useMemo(
      () => partialJSONParse(toolCall.function.arguments),
      [toolCall.function.arguments],
    );

    const toolName = toolCall.function.name;

    // Render based on status to preserve discriminated union type inference
    if (toolMessage) {
      return (
        <RenderComponent
          name={toolName}
          args={args}
          status={ToolCallStatus.Complete}
          result={toolMessage.content}
        />
      );
    } else if (isExecuting) {
      return (
        <RenderComponent
          name={toolName}
          args={args}
          status={ToolCallStatus.Executing}
          result={undefined}
        />
      );
    } else {
      return (
        <RenderComponent
          name={toolName}
          args={args}
          status={ToolCallStatus.InProgress}
          result={undefined}
        />
      );
    }
  },
  // Custom comparison function to prevent re-renders when tool call data hasn't changed
  ....
)

注意上面的RenderComponent

<RenderComponent
   name={toolName}
   args={args}
   status={ToolCallStatus.Complete}
   result={toolMessage.content}
   />

是经过useFrontendTool 注册的组件是的render

非controlled Generative UI 渲染逻辑

第二类和第三类的渲染逻辑走 role: “activity” 的渲染逻辑

1	const { renderActivityMessage } = useRenderActivityMessage();

 ----------------------- useRenderActivityMessage --------------------------------

const { copilotkit } = useCopilotKit();
const config = useCopilotChatConfiguration();
const agentId = config?.agentId ?? DEFAULT_AGENT_ID;

const renderers = copilotkit.renderActivityMessages;

// Find the renderer for a given activity type
  const findRenderer = useCallback(
    (activityType: string): ReactActivityMessageRenderer<unknown> | null => {
      if (!renderers.length) {
        return null;
      }

      const matches = renderers.filter(
        (renderer) => renderer.activityType === activityType,
      );

      return (
        matches.find((candidate) => candidate.agentId === agentId) ??
        matches.find((candidate) => candidate.agentId === undefined) ??
        renderers.find((candidate) => candidate.activityType === "*") ??
        null
      );
    },
    [agentId, renderers],
  );

注册时的处理第2类和第3类的render

a2UI

dynamic schema UI 渲染messages

fixed schema UI 渲染messages

/**
 * Renders a single A2UI surface using the React renderer.
 * Wraps A2UIProvider + A2UIRenderer and bridges actions back to CopilotKit.
 */
function ReactSurfaceHost({
  surfaceId,
  operations,
  theme,
  agent,
  copilotkit,
  catalog,
}: ReactSurfaceHostProps) {
  // Bridge: when the React renderer dispatches an action, forward to CopilotKit
  const handleAction = useCallback(
    async (message: A2UIClientEventMessage) => {
      if (!agent) return;

      const action = message.userAction as A2UIUserAction | undefined;

      try {
        copilotkit.setProperties({
          ...(copilotkit.properties ?? {}),
          a2uiAction: message,
        });

        await copilotkit.runAgent({ agent });
      } finally {
        if (copilotkit.properties) {
          const { a2uiAction, ...rest } = copilotkit.properties;
          copilotkit.setProperties(rest);
        }
      }
    },
    [agent, copilotkit],
  );

  return (
    <div className="cpk:flex cpk:w-full cpk:flex-none cpk:flex-col cpk:gap-4">
      <A2UIProvider onAction={handleAction} theme={theme} catalog={catalog}>
        <SurfaceMessageProcessor
          surfaceId={surfaceId}
          operations={operations}
        />
        <A2UISurfaceOrError surfaceId={surfaceId} />
      </A2UIProvider>
    </div>
  );
}

----------------------------SurfaceMessageProcessor---------------------------------------------

function SurfaceMessageProcessor({
  surfaceId,
  operations,
}: {
  surfaceId: string;
  operations: any[];
}) {
  const { processMessages, getSurface } = useA2UIActions();
  const lastHashRef = useRef<string>("");
  useEffect(() => {
    // Skip if operations haven't actually changed (deep compare via hash).
    // ACTIVITY_DELTA + ACTIVITY_SNAPSHOT can trigger multiple renders with
    // the same logical content but different object references.
    const hash = JSON.stringify(operations);
    if (hash === lastHashRef.current) return;
    lastHashRef.current = hash;

    // Filter out createSurface if the surface already exists — the
    // MessageProcessor throws on duplicate createSurface, but content
    // snapshots always include the full operation list.
    const existing = getSurface(surfaceId);
    const ops = existing
      ? operations.filter((op) => !op?.createSurface)
      : operations;

    // Error handling is done inside A2UIProvider.processMessages
    processMessages(ops);
  }, [processMessages, getSurface, surfaceId, operations]);

  return null;
}

processMessages 通过执行不同的操作实现a2UI的渲染

processCreateSurfaceMessage(message) {
    const payload = message.createSurface;
    const { surfaceId, catalogId, theme, sendDataModel } = payload;
    // Find catalog
    const catalog = this.catalogs.find(c => c.id === catalogId);
    if (!catalog) {
        throw new A2uiStateError(`Catalog not found: ${catalogId}`);
    }
    if (this.model.getSurface(surfaceId)) {
        throw new A2uiStateError(`Surface ${surfaceId} already exists.`);
    }
    const surface = new SurfaceModel(surfaceId, catalog, theme, sendDataModel ?? false);
    this.model.addSurface(surface);
}

processUpdateComponentsMessage(message) {
    const payload = message.updateComponents;
    if (!payload.surfaceId)
        return;
    const surface = this.model.getSurface(payload.surfaceId);
    if (!surface) {
        throw new A2uiStateError(`Surface not found for message: ${payload.surfaceId}`);
    }
    for (const comp of payload.components) {
        const { id, component, ...properties } = comp;
        if (!id) {
            throw new A2uiValidationError(`Component '${component}' is missing an 'id'.`);
        }
        const existing = surface.componentsModel.get(id);
        if (existing) {
            if (component && component !== existing.type) {
                // Recreate component if type changes
                surface.componentsModel.removeComponent(id);
                const newComponent = new ComponentModel(id, component, properties);
                surface.componentsModel.addComponent(newComponent);
            }
            else {
                existing.properties = properties;
            }
        }
        else {
            if (!component) {
                throw new A2uiValidationError(`Cannot create component ${id} without a type.`);
            }
            const newComponent = new ComponentModel(id, component, properties);
            surface.componentsModel.addComponent(newComponent);
        }
    }
}

mcp 渲染逻辑

详见下面代码，主要逻辑就是根据mcp配置通过 agent 发起请求获取到html 再通过通知方式将内容填到创建好的iframe中

/**
 * MCP Apps Extension Activity Renderer
 *
 * Renders MCP Apps UI in a sandboxed iframe with full protocol support.
 * Fetches resource content on-demand via proxied MCP requests.
 */
export const MCPAppsActivityRenderer: React.FC<MCPAppsActivityRendererProps> =
  function MCPAppsActivityRenderer({ content, agent }) {
    const containerRef = useRef<HTMLDivElement>(null);
    const iframeRef = useRef<HTMLIFrameElement | null>(null);
    const [iframeReady, setIframeReady] = useState(false);
    const [error, setError] = useState<Error | null>(null);
    const [isLoading, setIsLoading] = useState(true);
    const [iframeSize, setIframeSize] = useState<{
      width?: number;
      height?: number;
    }>({});
    const [fetchedResource, setFetchedResource] =
      useState<FetchedResource | null>(null);

    // Use refs for values that shouldn't trigger re-renders but need latest values
    const contentRef = useRef(content);
    contentRef.current = content;

    // Store agent in a ref for use in async handlers
    const agentRef = useRef(agent);
    agentRef.current = agent;

    // Ref to track fetch state - survives StrictMode remounts
    const fetchStateRef = useRef<{
      inProgress: boolean;
      promise: Promise<FetchedResource | null> | null;
      resourceUri: string | null;
    }>({ inProgress: false, promise: null, resourceUri: null });

    // Callback to send a message to the iframe
    const sendToIframe = useCallback((msg: JSONRPCMessage) => {
      if (iframeRef.current?.contentWindow) {
        console.log("[MCPAppsRenderer] Sending to iframe:", msg);
        iframeRef.current.contentWindow.postMessage(msg, "*");
      }
    }, []);

    // Callback to send a JSON-RPC response
    const sendResponse = useCallback(
      (id: string | number, result: unknown) => {
        sendToIframe({
          jsonrpc: "2.0",
          id,
          result,
        });
      },
      [sendToIframe],
    );

    // Callback to send a JSON-RPC error response
    const sendErrorResponse = useCallback(
      (id: string | number, code: number, message: string) => {
        sendToIframe({
          jsonrpc: "2.0",
          id,
          error: { code, message },
        });
      },
      [sendToIframe],
    );

    // Callback to send a notification
    const sendNotification = useCallback(
      (method: string, params?: Record<string, unknown>) => {
        sendToIframe({
          jsonrpc: "2.0",
          method,
          params: params || {},
        });
      },
      [sendToIframe],
    );

    // Effect 0: Fetch the resource content on mount
    // Uses ref-based deduplication to handle React StrictMode double-mounting
    useEffect(() => {
      const { resourceUri, serverHash, serverId } = content;

      // Check if we already have a fetch in progress for this resource
      // This handles StrictMode double-mounting - second mount reuses first mount's promise
      if (
        fetchStateRef.current.inProgress &&
        fetchStateRef.current.resourceUri === resourceUri
      ) {
        // Reuse the existing promise
        fetchStateRef.current.promise
          ?.then((resource) => {
            if (resource) {
              setFetchedResource(resource);
              setIsLoading(false);
            }
          })
          .catch((err) => {
            setError(err instanceof Error ? err : new Error(String(err)));
            setIsLoading(false);
          });
        return;
      }

      if (!agent) {
        setError(new Error("No agent available to fetch resource"));
        setIsLoading(false);
        return;
      }

      // Mark fetch as in progress
      fetchStateRef.current.inProgress = true;
      fetchStateRef.current.resourceUri = resourceUri;

      // Create the fetch promise using the queue to serialize requests
      const fetchPromise = (async (): Promise<FetchedResource | null> => {
        try {
          // Use queue to wait for agent to be idle and serialize requests
          const runResult = await mcpAppsRequestQueue.enqueue(agent, () =>
            agent.runAgent({
              forwardedProps: {
                __proxiedMCPRequest: {
                  serverHash,
                  serverId, // optional, takes precedence if provided
                  method: "resources/read",
                  params: { uri: resourceUri },
                },
              },
            }),
          );

          // Extract resource from result
          // The response format is: { contents: [{ uri, mimeType, text?, blob?, _meta? }] }
          const resultData = runResult.result as
            | { contents?: FetchedResource[] }
            | undefined;
          const resource = resultData?.contents?.[0];

          if (!resource) {
            throw new Error("No resource content in response");
          }

          return resource;
        } catch (err) {
          console.error("[MCPAppsRenderer] Failed to fetch resource:", err);
          throw err;
        } finally {
          // Mark fetch as complete
          fetchStateRef.current.inProgress = false;
        }
      })();

      // Store the promise for potential reuse
      fetchStateRef.current.promise = fetchPromise;

      // Handle the result
      fetchPromise
        .then((resource) => {
          if (resource) {
            setFetchedResource(resource);
            setIsLoading(false);
          }
        })
        .catch((err) => {
          setError(err instanceof Error ? err : new Error(String(err)));
          setIsLoading(false);
        });

      // No cleanup needed - we want the fetch to complete even if StrictMode unmounts
    }, [agent, content]);

    // Effect 1: Setup sandbox proxy iframe and communication (after resource is fetched)
    useEffect(() => {
      // Wait for resource to be fetched
      if (isLoading || !fetchedResource) {
        return;
      }

      // Capture container reference at effect start (refs are cleared during unmount)
      const container = containerRef.current;
      if (!container) {
        return;
      }

      let mounted = true;
      let messageHandler: ((event: MessageEvent) => void) | null = null;
      let initialListener: ((event: MessageEvent) => void) | null = null;
      let createdIframe: HTMLIFrameElement | null = null;

      const setup = async () => {
        try {
          // Create sandbox proxy iframe
          const iframe = document.createElement("iframe");
          createdIframe = iframe; // Track for cleanup
          iframe.style.width = "100%";
          iframe.style.height = "100px"; // Start small, will be resized by size-changed notification
          iframe.style.border = "none";
          iframe.style.backgroundColor = "transparent";
          iframe.style.display = "block";
          iframe.setAttribute(
            "sandbox",
            "allow-scripts allow-same-origin allow-forms",
          );

          // Wait for sandbox proxy to be ready
          const sandboxReady = new Promise<void>((resolve) => {
            initialListener = (event: MessageEvent) => {
              if (event.source === iframe.contentWindow) {
                if (
                  event.data?.method === "ui/notifications/sandbox-proxy-ready"
                ) {
                  if (initialListener) {
                    window.removeEventListener("message", initialListener);
                    initialListener = null;
                  }
                  resolve();
                }
              }
            };
            window.addEventListener("message", initialListener);
          });

          // Check mounted before adding to DOM (handles StrictMode double-mount)
          if (!mounted) {
            if (initialListener) {
              window.removeEventListener("message", initialListener);
              initialListener = null;
            }
            return;
          }

          // Build sandbox HTML with CSP domains from resource metadata
          const cspDomains = fetchedResource._meta?.ui?.csp?.resourceDomains;
          iframe.srcdoc = buildSandboxHTML(cspDomains);
          iframeRef.current = iframe;
          container.appendChild(iframe);

          // Wait for sandbox proxy to signal ready
          await sandboxReady;
          if (!mounted) return;

          console.log("[MCPAppsRenderer] Sandbox proxy ready");

          // Setup message handler for JSON-RPC messages from the inner iframe
          messageHandler = async (event: MessageEvent) => {
            if (event.source !== iframe.contentWindow) return;

            const msg = event.data as JSONRPCMessage;
            if (!msg || typeof msg !== "object" || msg.jsonrpc !== "2.0")
              return;

            console.log("[MCPAppsRenderer] Received from iframe:", msg);

            // Handle requests (need response)
            if (isRequest(msg)) {
              switch (msg.method) {
                case "ui/initialize": {
                  // Respond with host capabilities
                  sendResponse(msg.id, {
                    protocolVersion: PROTOCOL_VERSION,
                    hostInfo: {
                      name: "CopilotKit MCP Apps Host",
                      version: "1.0.0",
                    },
                    hostCapabilities: {
                      openLinks: {},
                      logging: {},
                    },
                    hostContext: {
                      theme: "light",
                      platform: "web",
                    },
                  });
                  break;
                }

                case "ui/message": {
                  // Add message to CopilotKit chat
                  const currentAgent = agentRef.current;

                  if (!currentAgent) {
                    console.warn(
                      "[MCPAppsRenderer] ui/message: No agent available",
                    );
                    sendResponse(msg.id, { isError: false });
                    break;
                  }

                  try {
                    const params = msg.params as {
                      role?: string;
                      content?: Array<{ type: string; text?: string }>;
                    };

                    // Extract text content from the message
                    const textContent =
                      params.content
                        ?.filter((c) => c.type === "text" && c.text)
                        .map((c) => c.text)
                        .join("\n") || "";

                    if (textContent) {
                      currentAgent.addMessage({
                        id: crypto.randomUUID(),
                        role: (params.role as "user" | "assistant") || "user",
                        content: textContent,
                      });
                    }
                    sendResponse(msg.id, { isError: false });
                  } catch (err) {
                    console.error("[MCPAppsRenderer] ui/message error:", err);
                    sendResponse(msg.id, { isError: true });
                  }
                  break;
                }

                case "ui/open-link": {
                  // Open URL in new tab
                  const url = msg.params?.url as string | undefined;
                  if (url) {
                    window.open(url, "_blank", "noopener,noreferrer");
                    sendResponse(msg.id, { isError: false });
                  } else {
                    sendErrorResponse(msg.id, -32602, "Missing url parameter");
                  }
                  break;
                }

                case "tools/call": {
                  // Proxy tool call to MCP server via agent.runAgent()
                  const { serverHash, serverId } = contentRef.current;
                  const currentAgent = agentRef.current;

                  if (!serverHash) {
                    sendErrorResponse(
                      msg.id,
                      -32603,
                      "No server hash available for proxying",
                    );
                    break;
                  }

                  if (!currentAgent) {
                    sendErrorResponse(
                      msg.id,
                      -32603,
                      "No agent available for proxying",
                    );
                    break;
                  }

                  try {
                    // Use queue to wait for agent to be idle and serialize requests
                    const runResult = await mcpAppsRequestQueue.enqueue(
                      currentAgent,
                      () =>
                        currentAgent.runAgent({
                          forwardedProps: {
                            __proxiedMCPRequest: {
                              serverHash,
                              serverId, // optional, takes precedence if provided
                              method: "tools/call",
                              params: msg.params,
                            },
                          },
                        }),
                    );

                    // The result from runAgent contains the MCP response
                    sendResponse(msg.id, runResult.result || {});
                  } catch (err) {
                    console.error("[MCPAppsRenderer] tools/call error:", err);
                    sendErrorResponse(msg.id, -32603, String(err));
                  }
                  break;
                }

                default:
                  sendErrorResponse(
                    msg.id,
                    -32601,
                    `Method not found: ${msg.method}`,
                  );
              }
            }

            // Handle notifications (no response needed)
            if (isNotification(msg)) {
              switch (msg.method) {
                case "ui/notifications/initialized": {
                  console.log("[MCPAppsRenderer] Inner iframe initialized");
                  if (mounted) {
                    setIframeReady(true);
                  }
                  break;
                }

                case "ui/notifications/size-changed": {
                  const { width, height } = msg.params || {};
                  console.log("[MCPAppsRenderer] Size change:", {
                    width,
                    height,
                  });
                  if (mounted) {
                    setIframeSize({
                      width: typeof width === "number" ? width : undefined,
                      height: typeof height === "number" ? height : undefined,
                    });
                  }
                  break;
                }

                case "notifications/message": {
                  // Logging notification from the app
                  console.log("[MCPAppsRenderer] App log:", msg.params);
                  break;
                }
              }
            }
          };

          window.addEventListener("message", messageHandler);

          // Extract HTML content from fetched resource
          let html: string;
          if (fetchedResource.text) {
            html = fetchedResource.text;
          } else if (fetchedResource.blob) {
            html = atob(fetchedResource.blob);
          } else {
            throw new Error("Resource has no text or blob content");
          }

          // Send the resource content to the sandbox proxy
          sendNotification("ui/notifications/sandbox-resource-ready", { html });
        } catch (err) {
          console.error("[MCPAppsRenderer] Setup error:", err);
          if (mounted) {
            setError(err instanceof Error ? err : new Error(String(err)));
          }
        }
      };

      setup();

      return () => {
        mounted = false;
        // Clean up initial listener if still active
        if (initialListener) {
          window.removeEventListener("message", initialListener);
          initialListener = null;
        }
        if (messageHandler) {
          window.removeEventListener("message", messageHandler);
        }
        // Remove the iframe we created (using tracked reference, not DOM query)
        // This works even if containerRef.current is null during unmount
        if (createdIframe) {
          createdIframe.remove();
          createdIframe = null;
        }
        iframeRef.current = null;
      };
    }, [
      isLoading,
      fetchedResource,
      sendNotification,
      sendResponse,
      sendErrorResponse,
    ]);

    // Effect 2: Update iframe size when it changes
    useEffect(() => {
      if (iframeRef.current) {
        if (iframeSize.width !== undefined) {
          // Use minWidth with min() to allow expansion but cap at 100%
          iframeRef.current.style.minWidth = `min(${iframeSize.width}px, 100%)`;
          iframeRef.current.style.width = "100%";
        }
        if (iframeSize.height !== undefined) {
          iframeRef.current.style.height = `${iframeSize.height}px`;
        }
      }
    }, [iframeSize]);

    // Effect 3: Send tool input when iframe ready
    useEffect(() => {
      if (iframeReady && content.toolInput) {
        console.log("[MCPAppsRenderer] Sending tool input:", content.toolInput);
        sendNotification("ui/notifications/tool-input", {
          arguments: content.toolInput,
        });
      }
    }, [iframeReady, content.toolInput, sendNotification]);

    // Effect 4: Send tool result when iframe ready
    useEffect(() => {
      if (iframeReady && content.result) {
        console.log("[MCPAppsRenderer] Sending tool result:", content.result);
        sendNotification("ui/notifications/tool-result", content.result);
      }
    }, [iframeReady, content.result, sendNotification]);

    // Determine border styling based on prefersBorder metadata from fetched resource
    // true = show border/background, false = none, undefined = host decides (we default to none)
    const prefersBorder = fetchedResource?._meta?.ui?.prefersBorder;
    const borderStyle =
      prefersBorder === true
        ? {
            borderRadius: "8px",
            backgroundColor: "#f9f9f9",
            border: "1px solid #e0e0e0",
          }
        : {};

    return (
      <div
        ref={containerRef}
        style={{
          width: "100%",
          height: iframeSize.height ? `${iframeSize.height}px` : "auto",
          minHeight: "100px",
          overflow: "hidden",
          position: "relative",
          ...borderStyle,
        }}
      >
        {isLoading && (
          <div style={{ padding: "1rem", color: "#666" }}>Loading...</div>
        )}
        {error && (
          <div style={{ color: "red", padding: "1rem" }}>
            Error: {error.message}
          </div>
        )}
      </div>
    );
  };

open Generative UI 渲染逻辑

全开放式渲染逻辑过程直接从后端agent content 中获取html 再通过构建沙河环境，将html 渲染到iframe中


function OpenGenerativeUIActivityRendererInner({ content }: InnerProps) {
   const initialHeight = content.initialHeight ?? 200;
   const [autoHeight, setAutoHeight] = useState<number | null>(null);
   const sandboxFunctions = useSandboxFunctions();

   const localApi = useMemo(() => {
     const api: Record<string, Function> = {};
     for (const fn of sandboxFunctions) {
       api[fn.name] = fn.handler;
     }
     return api;
   }, [sandboxFunctions]);

   // Join html chunks only when streaming is complete
   const fullHtml =
     content.htmlComplete && content.html?.length
       ? content.html.join("")
       : undefined;
   // CSS from the dedicated parameter (available once cssComplete)
   const css = content.cssComplete ? content.css : undefined;

   ....
}

CopilotRuntime

本质是一个代理, 或者代理适配器(多个agent存在时)，寻找后端agent 服务，并将前端请求转发给后端agent 服务

On the server, CopilotRuntime accepts a map of AG-UI AbstractAgent instances. A framework adapter, an HttpAgent pointing at a remote server, and a custom implementation all use the same request path:

The runtime resolves the target agent by ID.
It clones the agent for request isolation and supplies messages, state, and thread context.
AgentRunner executes the agent and receives AG-UI events.
The runtime encodes those events as SSE and streams them to the frontend proxy.
The backend framework can change without forcing a corresponding change to the frontend AG-UI contract.

官方文档

它是一个框架无关代理，所以也可以用在支持Fetch API的node 层运行时中
Deploy to any runtime

AGUI 协议

abstractAgent

[AbstractAgent Api] (https://docs.ag-ui.com/sdk/js/client/abstract-agent)
AbstractAgent 源码

abstract run(input: RunAgentInput): Observable<BaseEvent>;

public async runAgent(
    parameters?: RunAgentParameters,
    subscriber?: AgentSubscriber,
  ): Promise<RunAgentResult> {
    try {
      this.isRunning = true;
      this.agentId = this.agentId ?? uuidv4();
      const input = this.prepareRunAgentInput(parameters);

      this.debugLogger?.lifecycle("LIFECYCLE", "Run started:", {
        agentId: this.agentId,
        threadId: this.threadId,
      });

      let result: any = undefined;
      const currentMessageIds = new Set(this.messages.map((message) => message.id));

      const subscribers: AgentSubscriber[] = [
        {
          onRunFinishedEvent: (params) => {
            if (params.outcome === "success") {
              result = params.result;
            }
          },
        },
        ...this.subscribers,
        subscriber ?? {},
      ];

      await this.onInitialize(input, subscribers);

      // Per-run detachment signal + completion promise
      this.activeRunDetach$ = new Subject<void>();
      let resolveActiveRunCompletion: (() => void) | undefined;
      this.activeRunCompletionPromise = new Promise<void>((resolve) => {
        resolveActiveRunCompletion = resolve;
      });

      // 没有中间件的话直接运行run 有种中间件的话，会先运行中间件，初始值是agent本身，reduceRight 到最后是会执行agent.run(input) 方法

      const pipeline = pipe(
        () => {
          // Build middleware chain using reduceRight so middlewares can intercept runs.
          if (this.middlewares.length === 0) {
            return this.run(input);
          }

          const chainedAgent = this.middlewares.reduceRight(
            (nextAgent: AbstractAgent, middleware) =>
              ({
                run: (i: RunAgentInput) => middleware.run(i, nextAgent),
                get messages() {
                  return nextAgent.messages;
                },
                get state() {
                  return nextAgent.state;
                },
              }) as AbstractAgent,
            this, // Original agent is the final 'next'
          );

          return chainedAgent.run(input);
        },
        transformChunks(this.debugLogger),
        verifyEvents(this.debugLogger),
        // Stop processing immediately when this run is detached
        (source$) => source$.pipe(takeUntil(this.activeRunDetach$!)),
        (source$) => this.apply(input, source$, subscribers),
        (source$) => this.processApplyEvents(input, source$, subscribers),
        catchError((error) => {
          this.debugLogger?.lifecycle("LIFECYCLE", "Run errored:", {
            agentId: this.agentId,
            error: error instanceof Error ? error.message : String(error),
          });
          this.isRunning = false;
          return this.onError(input, error, subscribers);
        }),
        finalize(() => {
          this.debugLogger?.lifecycle("LIFECYCLE", "Run finished:", {
            agentId: this.agentId,
            threadId: this.threadId,
          });
          this.isRunning = false;
          void this.onFinalize(input, subscribers);
          resolveActiveRunCompletion?.();
          resolveActiveRunCompletion = undefined;
          this.activeRunCompletionPromise = undefined;
          this.activeRunDetach$ = undefined;
        }),
      );

      await lastValueFrom(pipeline(of(null)));
      const newMessages = structuredClone_(this.messages).filter(
        (message: Message) => !currentMessageIds.has(message.id),
      );
      return { result, newMessages };
    } finally {
      this.isRunning = false;
    }
  }

httpAgent

[HttpAgent Api] (https://docs.ag-ui.com/sdk/js/client/http-agent)

httpAgent 基于 abstractAgent 抽象类

HttpAgent 源码

httpAgent 会实现 run 函数，发起真正的http 请求


/**
   * Returns the fetch config for the http request.
   * Override this to customize the request.
   *
   * @returns The fetch config for the http request.
   */
  protected requestInit(input: RunAgentInput): RequestInit {
    return {
      method: "POST",
      headers: {
        ...this.headers,
        "Content-Type": "application/json",
        Accept: "text/event-stream",
      },
      body: JSON.stringify(input),
      signal: this.abortController.signal,
    };
  }
  run(input: RunAgentInput): Observable<BaseEvent> {
    const httpEvents = runHttpRequest(() => this.fetch(this.url, this.requestInit(input)));
    return transformHttpEventStream(httpEvents, this.debugLogger);
  }

runHttpRequest

runHttpRequest 处理流数据转成 HttpEventType 流


export enum HttpEventType {
  HEADERS = "headers",
  DATA = "data",
}

export const runHttpRequest = (
  fetchResponse: () => Promise<Response>,
): Observable<HttpEvent> => {
  // Defer the fetch so that it's executed when subscribed to
  return defer(() => from(fetchResponse())).pipe(
    switchMap((response) => {
      if (!response.ok) {
        const contentType = response.headers.get("content-type") || "";
        // Read the (small) error body once, then error the stream
        return from(response.text()).pipe(
          mergeMap((text) => {
            let payload: unknown = text;
            if (contentType.includes("application/json")) {
              try { payload = JSON.parse(text); } catch {/* keep raw text */}
            }
            const err: any = new Error(
              `HTTP ${response.status}: ${typeof payload === "string" ? payload : JSON.stringify(payload)}`
            );
            err.status = response.status;
            err.payload = payload;
            return throwError(() => err);
          })
        );
      }
      // Emit headers event first
      const headersEvent: HttpHeadersEvent = {
        type: HttpEventType.HEADERS,
        status: response.status,
        headers: response.headers,
      };

      const reader = response.body?.getReader();
      if (!reader) {
        return throwError(() => new Error("Failed to getReader() from response"));
      }

      return new Observable<HttpEvent>((subscriber) => {
        // Emit headers event first
        subscriber.next(headersEvent);

        (async () => {
          try {
            while (true) {
              const { done, value } = await reader.read();
              if (done) break;
              // Emit data event instead of raw Uint8Array
              const dataEvent: HttpDataEvent = {
                type: HttpEventType.DATA,
                data: value,
              };
              subscriber.next(dataEvent);
            }
            subscriber.complete();
          } catch (error) {
            subscriber.error(error);
          }
        })();

        return () => {
          reader.cancel().catch((error) => {
            if ((error as DOMException)?.name === "AbortError") {
              return;
            }

            throw error;
          });
        };
      });
    }),
  );
};

transformHttpEventStream

export enum EventType {
  TEXT_MESSAGE_START = "TEXT_MESSAGE_START",
  TEXT_MESSAGE_CONTENT = "TEXT_MESSAGE_CONTENT",
  TEXT_MESSAGE_END = "TEXT_MESSAGE_END",
  TEXT_MESSAGE_CHUNK = "TEXT_MESSAGE_CHUNK",
  TOOL_CALL_START = "TOOL_CALL_START",
  TOOL_CALL_ARGS = "TOOL_CALL_ARGS",
  TOOL_CALL_END = "TOOL_CALL_END",
  TOOL_CALL_CHUNK = "TOOL_CALL_CHUNK",
  TOOL_CALL_RESULT = "TOOL_CALL_RESULT",
  THINKING_TEXT_MESSAGE_END = "THINKING_TEXT_MESSAGE_END",
  STATE_SNAPSHOT = "STATE_SNAPSHOT",
  STATE_DELTA = "STATE_DELTA",
  MESSAGES_SNAPSHOT = "MESSAGES_SNAPSHOT",
  ACTIVITY_SNAPSHOT = "ACTIVITY_SNAPSHOT",
  ACTIVITY_DELTA = "ACTIVITY_DELTA",
  RAW = "RAW",
  CUSTOM = "CUSTOM",
  RUN_STARTED = "RUN_STARTED",
  RUN_FINISHED = "RUN_FINISHED",
  RUN_ERROR = "RUN_ERROR",
  STEP_STARTED = "STEP_STARTED",
  STEP_FINISHED = "STEP_FINISHED",
  REASONING_START = "REASONING_START",
  REASONING_MESSAGE_START = "REASONING_MESSAGE_START",
  REASONING_MESSAGE_CONTENT = "REASONING_MESSAGE_CONTENT",
  REASONING_MESSAGE_END = "REASONING_MESSAGE_END",
  REASONING_MESSAGE_CHUNK = "REASONING_MESSAGE_CHUNK",
  REASONING_END = "REASONING_END",
  REASONING_ENCRYPTED_VALUE = "REASONING_ENCRYPTED_VALUE",
}

/**
 * Transforms HTTP events into BaseEvents using the appropriate format parser based on content type.
 */

export const transformHttpEventStream = (
  source$: Observable<HttpEvent>,
  debugLogger?: DebugLoggerInput,
): Observable<BaseEvent> => {

  .....

   // If we get headers and haven't initialized a parser yet, check content type
      if (event.type === HttpEventType.HEADERS && !parserInitialized) {
        parserInitialized = true;
        const contentType = event.headers.get("content-type");

        log?.lifecycle("HTTP", "Stream format detected:", {
          contentType,
          parser: contentType === proto.AGUI_MEDIA_TYPE ? "protobuf" : "sse",
        });

        // Choose parser based on content type
        if (contentType === proto.AGUI_MEDIA_TYPE) {
          // Use protocol buffer parser
          parseProtoStream(bufferSubject).subscribe({
            next: (event) => eventSubject.next(event),
            error: (err) => eventSubject.error(err),
            complete: () => eventSubject.complete(),
          });
        } else {
          // Use SSE JSON parser for all other cases
          parseSSEStream(bufferSubject, log).subscribe({
            next: (json) => {
              try {
                const parsedEvent = EventSchemas.parse(json);
                log?.event("HTTP", "Event validated:", parsedEvent, {
                  type: parsedEvent.type,
                  valid: true,
                });
                eventSubject.next(parsedEvent as BaseEvent);
              } catch (err) {
                log?.event("HTTP", "Event invalid:", { json, error: String(err) });
                eventSubject.error(err);
              }
            },
            error: (err) => {
              ....
              return eventSubject.error(err);
            },
            complete: () => eventSubject.complete(),
          });
        }
      } else if (!parserInitialized) {
        eventSubject.error(new Error("No headers event received before data events"));
      }

}

parseSSEStream

源码

/**
 * Parses a stream of HTTP events into a stream of JSON objects using Server-Sent Events (SSE) format.
 * Strictly follows the SSE standard where:
 * - Events are separated by double newlines ('\n\n')
 * - Only 'data:' prefixed lines are processed
 * - Multi-line data events are supported and joined
 * - Non-data fields (event, id, retry) are ignored
 */

 next: (event: HttpEvent) => {
      if (event.type === HttpEventType.HEADERS) {
        return;
      }

      if (event.type === HttpEventType.DATA && event.data) {
        // Decode chunk carefully to handle UTF-8
        const text = decoder.decode(event.data, { stream: true });
        buffer += text;

        // Process complete events (separated by double newlines)
        const events = buffer.split(/\n\n/);
        // Keep the last potentially incomplete event in buffer
        buffer = events.pop() || "";

        for (const event of events) {
          processSSEEvent(event);
        }
      }
    },

/**
   * Helper function to process an SSE event.
   * Extracts and joins data lines, then parses the result as JSON.
   *
   * Follows the SSE spec by processing lines starting with 'data:',
   * ignoring a single space if it is present after the colon.
   *
   * @param eventText The raw event text to process
   */
  function processSSEEvent(eventText: string) {
    const lines = eventText.split("\n");
    const dataLines: string[] = [];

    for (const line of lines) {
      if (line.startsWith("data:")) {
        // Remove 'data:' prefix, and optionally a single space afterwards
        dataLines.push(line.slice(5).replace(/^ /, ""));
      }
    }

    // Only process if we have data lines
    if (dataLines.length > 0) {
      try {
        // Join multi-line data and parse JSON
        const jsonStr = dataLines.join("\n");
        const json = JSON.parse(jsonStr);
        log?.event("SSE", "Event received:", json, { type: json.type });
        jsonSubject.next(json);
      } catch (err) {
        jsonSubject.error(err);
      }
    }
  }

parseProtoStream

源码

/**
 * Parses a stream of HTTP events into a stream of BaseEvent objects using Protocol Buffer format.
 * Each message is prefixed with a 4-byte length header (uint32 in big-endian format)
 * followed by the protocol buffer encoded message.
 */

   next: (event: HttpEvent) => {
      if (event.type === HttpEventType.HEADERS) {
        return;
      }

      if (event.type === HttpEventType.DATA && event.data) {
        // Append the new data to our buffer
        const newBuffer = new Uint8Array(buffer.length + event.data.length);
        newBuffer.set(buffer, 0);
        newBuffer.set(event.data, buffer.length);
        buffer = newBuffer;

        // Process as many complete messages as possible
        processBuffer();
      }
    },

 /**
   * Process as many complete messages as possible from the buffer
   */
function processBuffer() {
    // Keep processing while we have enough data for at least a header (4 bytes)
    while (buffer.length >= 4) {
      // Read message length from the first 4 bytes (big-endian uint32)
      const view = new DataView(buffer.buffer, buffer.byteOffset, 4);
      const messageLength = view.getUint32(0, false); // false = big-endian

      // Check if we have the complete message (header + message body)
      const totalLength = 4 + messageLength;
      if (buffer.length < totalLength) {
        // Not enough data yet, wait for more
        break;
      }

      try {
        // Extract the message (skipping the 4-byte header)
        const message = buffer.slice(4, totalLength);

        // Decode the protocol buffer message using the imported decode function
        const event = proto.decode(message);

        // Emit the parsed event
        eventSubject.next(event);

        // Remove the processed message from the buffer
        buffer = buffer.slice(totalLength);
      } catch (error: unknown) {
        const errorMessage = error instanceof Error ? error.message : String(error);
        eventSubject.error(new Error(`Failed to decode protocol buffer message: ${errorMessage}`));
        return;
      }
    }
  }

为什么默认 copilotKit 实例不同动作可以发送不同http path 请求

从官网文档中可以看出，默认的 copilotKit 实例不同动作会基于 basePath 发送不同的http path 请求
实际请求时确实从寻找agent, 链接agent 服务，到向agent 提问都发送不同的http path 请求

原因：
useAgent() 函数会返回一个 ProxiedCopilotRuntimeAgent 对象
ProxiedCopilotRuntimeAgent 从httpAgent继承，
ProxiedCopilotRuntimeAgent 这个类会重新生成run 的请求url, 重写 connect 方法过程会指定，

/**
 * ProxiedCopilotRuntimeAgent 从httpAgent继承
 * 
 */


super({
  ...config,
  url: runUrl,
});

/**   run  ********************/

public run(input: RunAgentInput): Observable<BaseEvent> {
    if (
      this.runtimeMode === "pending" ||
      (this.transport === "auto" &&
        this.runtimeMode !== RUNTIME_MODE_INTELLIGENCE)
    ) {
      return defer(() => from(this.ensureRuntimeConfiguration())).pipe(
        switchMap(() => this.run(input)),
      );
    }
    if (this.runtimeMode === RUNTIME_MODE_INTELLIGENCE) {
      return this.#runViaDelegate(input);
    }
    return this.#runViaHttp(input);
  }

 #runViaDelegate(input: RunAgentInput): Observable<BaseEvent> {
    return defer(() => from(this.resolveDelegate())).pipe(
      switchMap((delegate) => withAbortErrorHandling(delegate.run(input))),
    );
  }

  #runViaHttp(input: RunAgentInput): Observable<BaseEvent> {
    if (this.transport === "single") {
      if (!this.singleEndpointUrl) {
        throw new Error("Single endpoint transport requires a runtimeUrl");
      }

      const requestInit = this.createSingleRouteRequestInit(
        input,
        "agent/run",
        {
          agentId: this.routedAgentId(),
        },
      );
      const httpEvents = runHttpRequest(() =>
        this.fetch(this.singleEndpointUrl!, requestInit),
      );
      return withAbortErrorHandling(transformHttpEventStream(httpEvents));
    }
    /**
     * POST /api/copilotkit/agent/:agentId/run
     */
    return withAbortErrorHandling(super.run(input));
  }


/**
 * 
 * /api/copilotkit/info 
 */


private async fetchRuntimeInfo(): Promise<RuntimeInfo> {
    const headers: Record<string, string> = {
      ...this.headers,
    };

    if (this.transport === "auto") {
      return this.fetchRuntimeInfoAutoDetect(headers);
    }

    let init: RequestInit;
    let url: string;

    if (this.transport === "single") {
      if (!this.singleEndpointUrl) {
        throw new Error("Single endpoint transport requires a runtimeUrl");
      }
      if (!headers["Content-Type"]) {
        headers["Content-Type"] = "application/json";
      }
      url = this.runtimeUrl!;
      init = { method: "POST", body: JSON.stringify({ method: "info" }) };
    } else {
      url = `${this.runtimeUrl}/info`;
      init = {};
    }
    ....

}

ProxiedCopilotRuntimeAgent 源码

Anthropic

发表于 2026-07-30

基本使用

from anthropic import Anthropic
from helper import load_env
load_env()

client = Anthropic()
MODEL_NAME="claude-3-5-sonnet-20241022"

response = client.messages.create(
    model=MODEL_NAME,
    max_tokens=1000,
    messages=[
        {"role": "user", "content": "Write a haiku about Anthropic"}
    ]
)

print(response.content[0].text)

返回结果

content: 消息的内容，可以是文本、图片、文件等
model: 当前使用的模型名称
role: 消息的角色，user 或 assistant assistant 消息会被自动加入，我们一般不需要自己构造
stop_reason: 停止生成的原因，可以是max_tokens、max_seconds、stop_sequence等
input_tokens: 输入的token数
output_tokens: 输出的token数

参数

stop_sequence: 停止生成的序列，eg. \n 如果client.messages.create 有配置，当生成的内容包含 \n 时，会停止生成

max_tokens: 生成最大长度，如果client.messages.create 有配置，当生成的内容超过 max_tokens 时，会停止生成

temperature: 生成结果的多样性。取值 0~1 之间，越大越散，越小越收敛

messages: 对话历史，包含用户和助手的消息，eg. [{“role”: “user”, “content”: “Write a haiku about Anthropic”}]

// 手动制造多轮对话（历史问答） 或者 Prefilling the Assistant Response （预填充助手回复）
messages=[
        {"role": "user", "content": "Hello! Only speak to me in Spanish"},
        {"role": "assistant", "content": "Hola!"},
        {"role": "user", "content": "How are you?"}
    ]
// chat bot 通过循环每次自动将历史问答加入到messages 中，实现多轮对话
print("Simple Chatbot (type 'quit' to exit)")
# Store conversation history
messages = []
while True:
    # Get user input
    user_input = input("You: ")
    # Check for quit command
    if user_input.lower() == 'quit':
        print("Goodbye!")
        break
    # Add user message to history
    messages.append({"role": "user", "content": user_input})
    try:
        # Get response from Claude
        response = client.messages.create(
            model=MODEL_NAME,
            max_tokens=200,
            messages=messages
        )
        # Extract and print Claude's response
        asst_message = response.content[0].text
        print("Assistant:", asst_message)
        
        # Add assistant response to history
        messages.append({"role": "assistant", "content": asst_message})
        
    except Exception as e:
        print(f"An error occurred: {e}")

多模态输入

messages 数组中 content 上面都是简单的文本，通过进行类型区分和扩展，可以进行多模态输入
此时content可以传入一个列表，每个元素是一个字典，包含type 和相应type 的数据字段

messages = [
    {
        "role": "user",
        "content": [{
            "type": "image", // 图片做输入 图片内容块
            "source": {
                "type": "base64", // 图片以base64 编码 输入
                "media_type": "image/png", // 图片类型
                "data": base64_string // 图片base64 字符串
            },
        },
        {
            "type": "text", // 文本做输入 文本内容块
            "text": """How many to-go containers of each type 
            are in this image?"""
        }]
    }
]

通过对图片进行输入，可以实现对图片进行识别和信息提取

import base64
import mimetypes

# 从图片路径创建图片内容块
def create_image_message(image_path):
    # Open the image file in "read binary" mode
    with open(image_path, "rb") as image_file:
        # Read the contents of the image as a bytes object
        binary_data = image_file.read()
    # Encode the binary data using Base64 encoding
    base64_encoded_data = base64.b64encode(binary_data)
    # Decode base64_encoded_data from bytes to a string
    base64_string = base64_encoded_data.decode('utf-8')
    # Get the MIME type of the image based on its file extension
    mime_type, _ = mimetypes.guess_type(image_path)
    # Create the image block
    image_block = {
        "type": "image",
        "source": {
            "type": "base64",
            "media_type": mime_type,
            "data": base64_string
        }
    }
    
    
    return image_block

# 创建messages
messages = [
    {
        "role": "user",
        "content": [
            create_image_message("./images/invoice.png"),
            {"type": "text", "text": """
                Generate a JSON object representing the contents # 提取图片中的信息，生成一个JSON对象
                of this invoice.  It should include all dates,
                dollar amounts, and addresses. 
                Only respond with the JSON itself.
            """
            }
        ]
    }
]

response = client.messages.create(
    model=MODEL_NAME,
    max_tokens=2048,
    messages=messages
)
print(response.content[0].text)

=====>
```json
{
  "invoice": {
    "invoice_number": "INV-2024-0042",
    "invoice_date": "March 17, 2025",
    "due_date": "April 16, 2025",
    "vendor": {
      "company_name": "ACME CORPORATION",
      "address": {
        "street": "123 Business Avenue",
        "city": "Silicon Valley",
        "state": "CA",
        "zip": "94025"
      },
      "email": "accounts@acmecorp.com"
    },
    "bill_to": {
      "address": {
        "street": "789 Market Street, Suite 500",
        "city": "Los Angeles",
        "state": "CA",
        "zip": "90015"
      }
    },
    ...
  }
}

流式输出

使用client.messages.stream 方法可以实现流式输出

with client.messages.stream(
    max_tokens=1024,
    messages=[{"role": "user", "content": "write a poem"}],
    model=MODEL_NAME,
) as stream:
  for text in stream.text_stream:
      print(text, end="", flush=True)

提示词

关键步骤

设置角色，背景，功能概述
制作模版
添加思考过程
阐述结果要求和注意事项
添加示例，包含尽可能多的场景，覆盖不同的输入情况

可以分步声明载组合起来

final_prompt = f"""
{setting_the_role}
{instruction_pt1}
{instruction_pt2}
{instruction_pt3}
"""

import re
def get_review_sentiment(review):
    #Insert the context into the prompt
    prompt = final_prompt.replace("{{CUSTOMER_REVIEW}}", review)
    # Send a request to Claude
    response = client.messages.create(
        model=MODEL_NAME,
        max_tokens=2000,
        messages=[
            {"role": "user", "content": prompt}        
        ]
    )
    output = response.content[0].text
    print("ENTIRE MODEL OUTPUT: ")
    print(output)
    
    sentiment = re.search(r'<json>(.*?)</json>', output, re.DOTALL) // 从模型输出中提取json
    
    if sentiment:
        print("FINAL JSON OUTPUT: ")
        print(sentiment.group(1).strip())
    else:
        print("No sentiment analysis in the response.")

review1 = """
I am in love with my Acme phone.  It's incredible.
It's a little expensive, but so worth it imo.
If you can afford it, it's worth it! 
I love the colors too!
"""

get_review_sentiment(review1)

prompt cache

通过在content block 中添加 cache_control 字段，可以控制是否使用缓存

当开启缓存时，当前content block 将被缓存，并在下次请求中作为缓存命中，从而节省计算资源。

开启缓存时写入要比普通token提问，要贵一些，但是在后续提问中会大幅提升响应速度

Cache write tokens are 25% more expensive than base input tokens
Cache read tokens are 90% cheaper than base input tokens
Regular input and output tokens are priced at standard rates

messages=[
    # ...long conversation so far
    {
        "role": "user",
        "content": [
            {
                "type": "text",
                "text": "Hello, can you tell me more about the solar system",
                "cache_control": {"type": "ephemeral"}
            }
        ]
    },
    {
        "role": "assistant",
        "content": "Certainly! The solar system is the collection of celestial bodies that orbit our Sun. It consists of eight planets, numerous moons, asteroids, comets, and other objects. The planets, in order from closest to farthest from the Sun, are: Mercury, Venus, Earth, Mars, Jupiter, Saturn, Uranus, and Neptune. Each planet has its own unique characteristics and features. Is there a specific aspect of the solar system you'd like to know more about?"
    },
    {
        "role": "user",
        "content": [
            {
                "type": "text",
                "text": "Tell me more about Mars.",
                "cache_control": {"type": "ephemeral"}
            }
        ]
    }
]

通过在多轮对话中一般会在每次提问时添加缓存设置，这样提问时就是写入缓存的过程，后续再提问时如果命中就是读缓存的过程，提高响应速度
缓存有ttl, 一般5分钟，如果缓存命中，则重新设置缓存时间
这样如果下一轮提问能命中当前提问的缓存，就可以直接从缓存中读取，快速获取到上一次的提问历史，进行速度提升，每一轮都这样就可以提升整体多轮会话速度

tool

tool schema 定义

{
    "name": "get_user",
    "description": "Looks up a user by email, phone, or username.",
    "input_schema": {
        "type": "object",
        "properties": {
            "key": {
                "type": "string",
                "enum": ["email", "phone", "username"],
                "description": "The attribute to search for a user by (email, phone, or username)."
            },
            "value": {
                "type": "string",
                "description": "The value to match for the specified attribute."
            }
        },
        "required": ["key", "value"]
    }
}

实际调用

def process_tool_call(tool_name, tool_input):
    if tool_name == "get_user":
        return db.get_user(tool_input["key"], tool_input["value"])
    elif tool_name == "get_order_by_id":
        return db.get_order_by_id(tool_input["order_id"])
    elif tool_name == "get_customer_orders":
        return db.get_customer_orders(tool_input["customer_id"])
    elif tool_name == "cancel_order":
        return db.cancel_order(tool_input["order_id"])

def simple_chat():
    system_prompt = """ xxxxxx """
    user_message = input("\nUser: ")
    messages = [{"role": "user", "content": user_message}]
    while True:
        if user_message == "quit":
            break
        #If the last message is from the assistant, 
        # get another input from the user
        if messages[-1].get("role") == "assistant":
            user_message = input("\nUser: ")
            messages.append({"role": "user", "content": user_message})

        #Send a request to Claude
        response = client.messages.create(
            model=MODEL_NAME,
            system=system_prompt,
            max_tokens=4096,
            tools=tools,
            messages=messages
        )
        # Update messages to include Claude's response
        messages.append(
            {"role": "assistant", "content": response.content}
        )

        #If Claude stops because it wants to use a tool:
        if response.stop_reason == "tool_use":
            #Naive approach assumes only 1 tool is called at a time
            tool_use = response.content[-1] 
            tool_name = tool_use.name
            tool_input = tool_use.input
            print(f"=====Claude wants to use the {tool_name} tool=====")


            #Actually run the underlying tool functionality on our db
            tool_result = process_tool_call(tool_name, tool_input)

            #Add our tool_result message:
            messages.append(
                {
                    "role": "user",
                    "content": [
                        {
                            "type": "tool_result",
                            "tool_use_id": tool_use.id,
                            "content": str(tool_result),
                        }
                    ],
                },
            )
        else: 
            #If Claude does NOT want to use a tool, 
            #just print out the text reponse
            model_reply = extract_reply(response.content[0].text)
            print("\nAcme Co Support: " + f"{model_reply}" )

CopilotKit 生成式UI框架

发表于 2026-07-29

本篇文章简要介绍基于 CopilotKit 生成式UI框架
主要用于解决 llm agent 响应结果的可视化展示问题
llm 产出如果不加以修饰绘制，用户看到的就只是普通文本，尤其对于一些数据类文本，缺少可视化能力
再丰富一些，可能会输出markdown格式进行一定规则的渲染，能力依然有限，不如前端组件渲染丰富
CopilotKit 框架基于AG-UI 协议，将前端展示组件库直接绑定到agent，作为tool 供agent 调使用，直接产生可视化展示结果

Generative UI

Render your agent’s state, progress, outputs, and tool calls with custom UI components in real-time. Bridges the gap between AI agents and user interfaces.

生成式UI就是丰富agent 响应结果的动态UI, 可以让agent 生成可交互式界面按照可生成范围大小可以分成三类

生成式UI 全控式，agent 只能在注册的指定几个组件范围内根据用户指令判断是否要使用，不能自定义组件
生成式UI 声明式，agent 可以在注册的一堆组件(相当于给agent一个组件库)中选取组件使用，也可以组装几个组件形成固定布局并进行注册，触发后直接根据数据渲染组件即可，也是不能自定义组件
生成式UI 开放式，agent 不注册任何组件，可以直接连接外部组件的mcp协议，实现组件的动态加载和使用，跨出了指定组件库的束缚，对于指定场景需求可以直接调用更复杂精细的匹配度更高的组件

下面CopilotKit 框架会按开放程度依次介绍

AG-UI 协议

AG-UI is an open, lightweight, event-based protocol that standardizes how AI agents connect to user-facing applications.

AG-UI is designed to be the general-purpose, bi-directional connection between a user-facing application and any agentic backend.

Built for simplicity and flexibility, it standardizes how agent state, UI intents, and user interactions flow between your model/agent runtime and user-facing frontend applications—to allow application developers to ship reliable, debuggable, user‑friendly agentic features fast while focusing on application needs and avoiding complex ad-hoc wiring.

AG-UI协议定义了agent直接和用户界面之间的通信协议，是三大开源代理协议之一，实现了agent 和应用页面之间实时，多模态，可交互用户体验
打破了现有前后开发模式(前端等待数据返回后，再进行渲染，用户再感知交互)

AG-UI协议出现的必要性

虽然代理只是软件，但它们展现出的特性使得传统的 REST/GraphQL API 难以为其提供服务：

代理运行时间长，并会流式传输中间工作——通常跨越多个会话。
代理具有非确定性，并且能够以非确定性的方式控制应用程序 UI。—>结果是不可预测的，需要前端进行实时渲染，需要自行选择渲染组件
代理同时混合结构化和非结构化 I/O（例如，文本和语音，以及工具调用和状态更新）。
代理需要用户交互式组合：例如，它们可能会调用子代理，而且通常是递归调用。

等等

它是一种基于事件的协议，它支持代理前端和后端之间的动态通信。它构建于 Web 的基础协议（HTTP、WebSocket）之上，作为一个专为代理时代设计的抽象层——弥合了传统客户端-服务器架构与 AI 代理的动态、有状态特性之间的差距
详见

区别其他协议

Layer	Protocol / Example	Purpose
Agent ⇔ User Interaction	AG-UI (Agent-User Interaction Protocol)	The open, event-based standard that connects agents to user-facing applications — enabling real-time, multimodal, interactive experiences.
Agent ⇔ Tools & Data	MCP (Model Context Protocol)	Open standard (originated by Anthropic) that lets agents securely connect to external systems — tools, workflows, and data sources.
Agent ⇔ Agent	A2A (Agent to Agent)	Open standard (originated by Google) which defines how agents coordinate and share work across distributed agentic systems.

区别 A2UI

A2UI is a generative UI specification - allowing agents to deliver UI widgets, where AG-UI is the Agent↔User Interaction protocol - which connects an agentic frontend to any agentic backend

A2UI, MCP-UI, and Open-JSON-UI are all generative UI specifications. Generative UIs allow agents to respond to users not only with text but also with dynamic UI components.

Specification	Origin / Maintainer	Purpose
A2UI	Google	A declarative, LLM-friendly Generative UI spec. JSONL-based and streaming, designed for platform-agnostic rendering.
Open-JSON-UI	OpenAI	An open standardization of OpenAI’s internal declarative Generative UI schema.
MCP-UI	Microsoft + Shopify	A fully open, iframe-based Generative UI standard extending MCP for user-facing experiences.

AG-UI is not a generative UI specification — it’s a User Interaction protocol that provides the bi-directional runtime connection between the agent and the application.

AG-UI natively supports all of the above generative UI specs and allows developers to define their own custom generative UI standards as well.

A2UI 是与agent进行交互的组件库的存在形式和定义规范

A2UI 是一种生成式 UI 规范 - 允许代理交付 UI 小部件，
AG-UI 是代理↔用户交互协议 - 它将代理前端连接到任何代理后端

AG-UI is a transport protocol connecting agent backends to frontends with real-time state sync. A2UI is a UI format: the payload that describes what to render. They’re complementary: use AG-UI as the pipe, A2UI as the content. AG-UI is a project by the CopilotKit team, who also contributed the A2UI Composer. AG-UI has day-zero A2UI compatibility.

CopilotKit 框架

一个可以链接多种前端框架和多种agent框架的生成式UI框架，利用AG-UI协议，实现agent和前端页面之间的实时通信渲染🔗

CopilotKit 中的生成式 UI 是一组基元，可让代理决定屏幕上显示的内容
从渲染您构建的特定应用程序组件，到根据目录组合布局，再到嵌入 MCP 服务器提供的沙盒 UI。

generative ui

详见

应用场景	方案	说明
让 Agent 渲染你已构建好的特定应用组件	Components as Tools（组件即工具）	将应用组件注册为前端工具；Agent 调用它时，CopilotKit 会内联渲染，并支持类型化的输入参数。
为 Agent 现有后端工具调用自定义 CopilotKit 渲染卡片样式	Tool Call Rendering（工具调用渲染）	将 Agent 已有的后端工具调用映射到自定义 UI 卡片，展示实时状态、参数和结果。
随 Agent 状态变化更新 UI（进度、草稿、仪表盘等）	State Rendering（状态渲染）	订阅 Agent 的流式状态，在值到达时重新渲染 UI。
在对话中展示模型的思考链	Reasoning（推理过程）	将模型的推理 token 作为一等消息类型内联渲染（默认卡片，或完全自定义）。
让 Agent 从你定义的组件目录中组合布局	A2UI：场景已知用 Fixed Schema（固定模式），场景未知用 Dynamic Schema（动态模式）	基于 Agent 输出的声明式 Schema 渲染 UI，通过你注册的组件目录进行组合。两种风格：Dynamic Schema（LLM 生成 Schema）和 Fixed Schema（你编写 Schema，Agent 提供数据）。
在对话中嵌入第三方 MCP 托管的 UI	MCP Apps（MCP 应用）	嵌入 MCP 服务器随其工具一同提供的 UI，在沙盒 iframe 中渲染，无需前端渲染器。

These primitives sit along a spectrum from author-controlled to agent-invented. Same frontend application, same runtime, same AG-UI protocol — the choice is per-feature, not per-product.

Controlled — you wrote the component; the agent only picks which one to render and what data to pass. Highest - predictability, highest engineering cost per capability. Components as Tools, Tool Call Rendering, State Rendering, Reasoning.

Declarative — the agent emits a structured spec; the frontend composes from a catalog you registered. Creativity inside a guardrail. A2UI (Dynamic and Fixed Schema).

Open-Ended — the UI is invented elsewhere (an MCP server) and you sandbox it. Highest expressive range, hardest to guarantee accessibility / brand / security. MCP Apps.

这些基本组件涵盖了从作者控制到代理创建的各个层面。相同的前端应用程序、相同的运行时环境、相同的 AG-UI 协议——选择是针对特定功能，而非针对特定产品。

受控型——您编写组件；代理仅负责选择要渲染的组件以及要传递的数据。可预测性最高，但每个功能的工程成本也最高。组件即工具、工具调用渲染、状态渲染、推理。

声明式——代理生成结构化的规范；前端根据您注册的目录进行组合。在一定的框架内进行创新。A2UI（动态和固定模式）。

开放式——用户界面在其他地方（MCP 服务器）创建，并由您进行沙盒化。表达范围最广，但最难保证可访问性/品牌/安全性。MCP 应用程序。

hooks

Hook	使用场景
useFrontendTool	给 Agent 提供一个客户端工具调用（在浏览器中运行逻辑），可选支持内联 UI。
useRenderTool	按名称（类型化）渲染特定工具调用的 UI，无需定义工具的处理函数。
useDefaultRenderTool	为没有特定渲染器的任意工具调用提供一个通用的通配渲染器。
useComponent	将一个 React 组件注册为命名工具渲染器（以组件优先的生成式 UI 方式）。
useHumanInTheLoop	在工具调用时暂停 Agent，等待用户批准、编辑或提供输入。
useInterrupt	处理 Agent 发起的中断，在用户响应后恢复执行。
useRenderToolCall	获取工具调用的渲染函数，用于驱动你自己的自定义聊天界面（无头模式）。

详见

多agent 切换

AG-UI：协议桥
CopilotKit 不会是一个代理框架中。运行时通过 AG-UI 与代理进行通信，AG-UI 是一种开放的事件驱动协议，可标准化代理与应用程序的通信方式：

事件驱动 — 16 种标准化事件类型（文本增量、工具调用、状态快照和增量、运行生命周期）从代理通过运行时流到前端。
双向——用户发送输入，代理响应，代理暂停以进行人机交互输入，前端公开代理可以调用的前端工具。
与传输无关——SSE、WebSockets、webhooks，无论您的堆栈喜欢什么。
与框架无关——每个受支持的集成都会附带一个精简的 AG-UI 适配器。使用一行运行时配置切换后端。
“代理的未来不是一家公司或一个平台，而是一个通过协议连接的代理生态系统。”

因为合约是一个协议，而不是 SDK 锁定，所以您可以交换代理层，而无需重写前端，并行运行多个代理后端，并与任何 AG-UI 兼容的内容集成：MCP 服务器、A2UI 组件、Oracle / Google / AWS 代理平台。

后端启动多服务

# 本脚本主要练习 
# 1. 声明不同模型agent 
# 2. 通过copilotKit 在 frontend 中直接进行切换调用不同agent

# 涉及前端文件
# 1. frontend/server.ts 在前端注册agent
# 2. frontend/src/main.tsx 在前端调用agent
# 3. frontend/src/App.tsx 在具体组件中切换指定agent

import warnings
from fastapi import FastAPI

# CopilotKit and AG-UI dependencies for the AG-UI server
from ag_ui_langgraph import add_langgraph_fastapi_endpoint
from copilotkit import LangGraphAGUIAgent
from copilotkit import CopilotKitMiddleware

# LangChain agent imports
from langchain.agents import create_agent
from langchain_openai import ChatOpenAI
from langgraph.checkpoint.memory import MemorySaver

#  create and start a Google ADK agent on port 8009
from ag_ui_adk import ADKAgent, add_adk_fastapi_endpoint
from google.adk.agents import LlmAgent

# Simple helper that starts the server and manages port conflicts
from helper import start_server
from helper import load_api_keys

from helper import get_ark_api_key
from helper import get_ark_api_base
from helper import get_ark_model_name

ARK_API_KEY = get_ark_api_key()
ARK_API_BASE = get_ark_api_base()
ARK_MODEL_NAME = get_ark_model_name()

load_api_keys()

warnings.filterwarnings("ignore")

# *************************** 建一个LangChain agent ***************************

# Build the AG-UI endpoint into a FastAPI app
# CopilotKit 通过兼容 AG-UI 的 HTTP 端点连接到agent。在这里，将启动一个 FastAPI 服务器并挂载一个 LangGraphAGUIAgent。
app = FastAPI()
graph = create_agent("openai:gpt-4.1")
agent = LangGraphAGUIAgent(
    name="lesson2_agent",
    description="Lesson 2 chart agent",
    graph=graph,
)
add_langgraph_fastapi_endpoint(app=app, agent=agent, path="/")

# Start the server
start_server(app, port=8002)

# Make a pre-built LangChain agent
# CopilotKitMiddleware connects your agent to CopilotKit — it lets the model discover and call frontend tools
# Without the middleware, the agent only sees backend-defined tools.
# CopilotKitMiddleware 是一个中间件，它允许您的代理发现并调用前端工具。
graph = create_agent(
    model=ChatOpenAI(model="gpt-4.1"),
    tools=[],
    middleware=[CopilotKitMiddleware()],
    checkpointer=MemorySaver(),
    system_prompt=("You are a helpful assistant"),
)

# Update the agent's graph (hot reloads) 
# 使用 agent.graph 替换graph
agent.graph = graph 
print("✓ Agent graph updated!")

# *************************** 建一个Google ADK agent ***************************

# create and start a Google ADK agent on port 8009
gemini_agent = LlmAgent(
    name="assistant",
    model="gemini-2.5-flash",
    instruction="Be helpful and fun!",
)

adk_agent = ADKAgent(
    adk_agent=gemini_agent,
    app_name="demo_app",
    user_id="demo_user",
    session_timeout_seconds=3600,
    use_in_memory_services=True,
)

app_adk = FastAPI()
add_adk_fastapi_endpoint(app_adk, adk_agent, path="/")

start_server(app_adk, port=8009)

#* ************************* 建一个火山引擎 ADK agent ***************************

ark_app_sdk = FastAPI()
ark_graph = create_agent(
    model=ChatOpenAI(
        openai_api_key=ARK_API_KEY,
        openai_api_base=ARK_API_BASE,
        model_name=ARK_MODEL_NAME, # 从环境变量读取模型
    ),
    tools=[],
    middleware=[CopilotKitMiddleware()],
    checkpointer=MemorySaver(),
    system_prompt=("You are a helpful assistant"),
)
ark_agent = LangGraphAGUIAgent(
    name="ark_agent",
    description="Ark agent for Lesson 2",
    graph=ark_graph,
)
add_langgraph_fastapi_endpoint(ark_app_sdk, ark_agent, path="/")

# Start the server
start_server(ark_app_sdk, port=8010)

# start_server 适合在 Jupyter Notebook 中运行（daemon 线程）
# 在 Notebook 里，Python 进程（Kernel）一直跑着，所以 daemon 线程也会一直活着
# 线程类型                     主程序退出时          适合场景 
# 守护线程 ( daemon=True )     ✅ 一起被杀掉         Jupyter Notebook、临时后台任务 
# 非守护线程 ( daemon=False )  ❌ 主程序会等它跑完     独立服务、常驻进程

# 下面的循环让终端运行时服务保持活跃
print("\n" + "="*50)
print("✅ 所有服务已启动，按 Ctrl+C 停止")
print("   Agent 服务: http://localhost:8002")
print("   ADK 服务:   http://localhost:8009")
print("   Ark 服务:   http://localhost:8010")
print("="*50)

import time
try:
    while True:
        time.sleep(60)
except KeyboardInterrupt:
    print("\n\n🛑 服务已停止")
    import sys
    sys.exit(0)

前端注册多个agent

import { HttpAgent } from "@ag-ui/client";
import { LangGraphHttpAgent } from "@copilotkit/runtime/langgraph";
import { CopilotRuntime, createCopilotEndpoint } from "@copilotkit/runtime/v2";
import { serve } from "@hono/node-server";

// 下面这三个agent 都指向后端agent 服务
const langGraphAgent = new LangGraphHttpAgent({
  url: process.env.LANGGRAPH_DEPLOYMENT_URL || "http://localhost:8002",
});

const adkAgent = new HttpAgent({
  url: process.env.ADK_AGENT_URL || "http://localhost:8009",
});

const arkAgent = new HttpAgent({
  url: process.env.ARK_AGENT_URL || "http://localhost:8010",
});

// The CopilotRuntime is the secure bridge between your frontend and agent backend
// 将不同的agent 注册到runtime 中 方便前端调用
const runtime = new CopilotRuntime({
  agents: {
    default: langGraphAgent, // Register  LangChain agent as the default agent
    gemini: adkAgent,
    ark: arkAgent,
  },
});

const app = createCopilotEndpoint({
  runtime,
  basePath: "/api/copilotkit",
});

serve({ fetch: app.fetch, port: 4002 }, () => {
  console.log("CopilotKit API server running at http://localhost:4002");
});

前端指定agent

import { CopilotChat } from "@copilotkit/react-core/v2";

export const agentId = "ark"; // 指定用哪个agent 上面注册的名称

export default function App() {
  return <CopilotChat agentId={agentId} />;
}

前端调用CopilotKit agent

import { StrictMode } from "react";
import { createRoot } from "react-dom/client";

import { CopilotKit } from "@copilotkit/react-core/v2";
import App from "./App";

import "./globals.css";
import "@copilotkit/react-core/v2/styles.css";

// 通过CopilotKit provider runtimeUrl 连接到agent
createRoot(document.getElementById("root")!).render(
  <StrictMode>
    <main className="h-screen w-screen">
      <CopilotKit runtimeUrl="/api/copilotkit" useSingleEndpoint={false}>
        <App />
      </CopilotKit>
    </main>
  </StrictMode>,
);

Controlled Generative UI 全控式

什么是受控生成式用户界面？
生成式用户界面是一种模式，在这种模式下，智能体会使用完全交互式的界面进行响应，而不仅仅是文本，受控生成式用户界面是最严格的形式：智能体只能渲染你明确注册过的组件。

其工作原理如下：
每个已注册的成分都作为一个工具被呈现出来，具有以下功能：
一个稳定的名字
一个格式化的输入模式
一个已映射的 React 组件

该代理不会生成随机的用户界面。它会将结构化数据传递给你已经构建的组件，而这些组件的定义是在前端阶段完成的，并且在运行时被注册到相应的位置。

优点

实施起来非常简单：只需注册一个组件即可。
视觉效果非常精致，因为所有渲染出的表面都是您亲手创作的。
强大的安全性保障：该模型只能调用那些具有有效参数的注册工具。
非常适合那些需要稳定性能的、高流量或关键任务型的用户界面场景。

缺点

随着新功能的不断增加，前端开发的工作量也在逐步上升：每一种模式都需要相应的组件来支持。
相比声明式或开放式生成式 UI，其表达自由度要低一些。

import { z } from "zod"

import { FlightCard, FlightCardProps } from "@/components/flight-card";
import { PieChart, PieChartProps } from "@/components/pie-chart";
import { useExampleSuggestions } from "@/hooks/use-example-suggestions";
import { CopilotChat } from "@copilotkit/react-core/v2";
import { useComponent } from "@copilotkit/react-core/v2";

/************ 使用useComponent 注册组件 ************ */
/*
 * 将react 组件直接注册为 a generative UI tool ，在响应中直接渲染组件
 * 注册组件[useComponent](https://docs.copilotkit.ai/generative-ui/your-components/display-only) 
registers a React component as a tool the agent can call inside `<CopilotChat />`. 
You define what's available; the agent picks when to use it.
useComponent({
  name: "component_name",
  description: "description for the agent to know about this component",
  parameters: z.object({ ... }),
  render: MyComponent,
});

- `name` (required `string`): tool name exposed to the model.
- `description` (optional `string`): tells the model when to call this tool.
- `parameters` (optional Zod schema): structured props passed in as arguments.
- `render` (required): a React component (rendered as `<Component {...args} />`), or a function receiving `{ args, status }` for custom rendering (loading states, wrappers, conditional display).
 */
export default function App() {

  // 🪁 Register a component that shows the name of the user
  useComponent({
    name: "showMyName",
    description: "Show the user's name in a card",
    parameters: z.object({ name: z.string() }),
    render: ({ name }) => <div className="bg-blue-500 p-4">Hi, {name}!</div>,
  });

  // 🪁 Resgister a pieChart component to show structured data
  useComponent({
    name: "pieChart",
    description: "Controlled Generative UI that displays data as a pie chart.",
    parameters: PieChartProps,
    render: PieChart,
  });

  // 🪁 Resgister a flightCard component to show flight data
  useComponent({
    name: "flightCard",
    description: "Controlled Generative UI that displays a single flight summary card.",
    parameters: FlightCardProps,
    render: FlightCard,
  });

  // 🪁 Add suggestions to our CopilotChat, will display through buttons
  useExampleSuggestions();

  return <CopilotChat />;

};

useComponent 注册组件

Tool-based Generative UI is the simplest form of Generative UI: you register a React component with useComponent, and CopilotKit exposes it to the agent as a tool. When the agent calls the tool, CopilotKit renders your component inline in the chat, passing the tool’s arguments straight through as typed props.

Unlike tool rendering, which wraps a real backend tool in a custom UI, tool-based GenUI is the component. There is no handler, no user interaction, no server-side execution. The agent decides when to show it, populates the data, and CopilotKit paints it.

详见

Deccontrolled Generative UI 声明式

什么是声明式生成式 UI？

声明式生成式 UI 让你定义一组 UI 构建块，由 Agent 将它们组合成界面。它从组件目录中选择组件，将它们排列成 Schema，并绑定运行时数据来填充最终结果。

它由三个部分协同工作：

组件目录：你的应用支持的 UI 基元，分为两部分：
- 定义（Definitions）：对每个组件的名称、属性和用途的平台无关描述。
- 渲染器（Renderers）：将定义转化为实际 UI 的平台特定实现（例如 React 组件）。
Schema：关于使用哪些组件、如何嵌套以及它们之间关系的结构化描述。
数据绑定（Data bindings）：运行时填充 Schema 的真实值，例如航班详情、指标或记录。

一个有用的心智模型是乐高积木：目录是装着积木的盒子，Schema 是积木拼接的方式，数据绑定在运行时填充最终细节。Agent 动态组装界面，而你的应用始终掌控一致性、安全性和渲染质量。

为什么使用声明式生成式 UI？

全控式生成式 UI 适用于流量最高、可预测性最重要的场景。但在长尾场景中（内部工具、边缘情况、多样化的用户目标），手动编写每个布局无法扩展。这正是声明式生成式 UI 的价值所在：Agent 从一组固定的构建块中组装 UI，在不牺牲安全性和一致性的前提下获得适应性。

优点

受约束的灵活性：Agent 在不超出你的组件体系的前提下适配界面。
自带组件：由你定义基元，Agent 决定如何组合它们。
每个场景工作量更少：只需定义一次目录，即可在各处复用。
原生跨平台：同一套 Schema 可在 Web、移动端、Slack 和短信中渲染。
比开放式生成式 UI 更省 Token：Agent 使用固定词汇表工作，无需生成任意代码。

缺点

像素级控制较弱：无法对最终界面进行精确微调。
可预测性较低：Agent 在相似场景下可能以不同方式组装组件。
更容易出错：Schema 和数据绑定可能以细微方式失败，需要验证和恢复逻辑。
需要前期设计：组件目录、Schema 格式和渲染器契约需要仔细定义。

开启a2ui 组件进行组件组装渲染

a2ui

通过启动A2UI可以将声明好的一系列组件打包成一个组件库供agent 使用
agent 可以根据prompt 和问题响应自行利用组件进行组装，生成对应的UI界面

import { serve } from "@hono/node-server";
import {
  CopilotRuntime,
  createCopilotEndpoint,
} from "@copilotkit/runtime/v2";
import { LangGraphHttpAgent } from "@copilotkit/runtime/langgraph";

const langGraphAgent = new LangGraphHttpAgent({ url: "http://localhost:8004" });

const runtime = new CopilotRuntime({
  agents: { default: langGraphAgent },
  a2ui: { injectA2UITool: true }, // 启动A2UI 组件 进行组件组装渲染
});

const app = createCopilotEndpoint({
  runtime,
  basePath: "/api/copilotkit",
});

serve({ fetch: app.fetch, port: 4004 }, () => {
  console.log("\u2713 CopilotKit API server running at http://localhost:4004");
});

当启用 A2UI 后，CopilotKit 会自动添加生成结构化 A2UI 输出所需的后端逻辑——您无需手动实现这一流程。

在底层，这一操作是通过两层工具调用来实现的：

当代理决定生成 A2UI 时，会触发一个外部工具调用。这样就能将与 A2UI 相关的逻辑与其他代理行为分离开来。

一个内部工具调用包含了结构化的 A2UI 负载。中间件会拦截这些参数，以实现流处理功能。

生成的输出结果也被记录在了代理的历史记录中，作为工具调用的结果之一。

dynamic schema UI

前端需要引入组件库 frontend/src/catalog 组件声明 + 组件实现 + 导出

A component definition is the contract for a primitive in your catalog:

a name the agent can reference (like Card, Row, or Text)
a schema for its props
and the React component that draws it in your app’s styles

// frontend/src/catalog/definitions.ts
import { z } from "zod";

export const demonstrationCatalogDefinitions = {
  Title: {
    description: "A heading. Use for section titles and page headers.",
    props: z.object({
      text: z.string(),
      level: z.string().optional(),
    }),
  },
  DashboardCard: {
    description:
      "A card container with title and optional subtitle. Has a 'child' slot for content (chart, metrics, etc). Use 'child' with a single component ID.",
    props: z.object({
      title: z.string(),
      subtitle: z.string().optional(),
      child: z.string().optional(),
    }),
  },
  Metric: {
    description:
      "A key metric display with label, value, and optional trend indicator. Great for KPIs and stats.",
    props: z.object({
      label: z.string(),
      value: z.string(),
      trend: z.enum(["up", "down", "neutral"]).optional(),
      trendValue: z.string().optional(),
    }),
  },

  PieChart: {
    description:
      "A pie/donut chart. Provide data as array of {label, value, color} objects.",
    props: z.object({
      data: z.array(
        z.object({
          label: z.string(),
          value: z.number(),
          color: z.string().optional(),
        }),
      ),
      innerRadius: z.number().optional(),
    }),
  },
  DataTable: {
    description: "A data table with columns and rows.",
    props: z.object({
      columns: z.array(z.object({ key: z.string(), label: z.string() })),
      rows: z.array(z.record(z.any())),
    }),
  }
};

/** Type helper for renderers */
export type DemonstrationCatalogDefinitions =
  typeof demonstrationCatalogDefinitions;

// frontend/src/catalog/renderers.tsx


import React from "react";
import {
  PieChart as RechartsPie,
  Pie,
  Cell,
  ResponsiveContainer,
  BarChart as RechartsBar,
  Bar,
  XAxis,
  YAxis,
  Tooltip,
  CartesianGrid,
} from "recharts";
import {
  createCatalog,
  type CatalogRenderers,
} from "@copilotkit/a2ui-renderer";
import {
  demonstrationCatalogDefinitions,
  type DemonstrationCatalogDefinitions,
} from "./definitions";

function resolveText(value: unknown): string {
  if (typeof value === "string") return value;
  if (value && typeof value === "object" && "path" in value)
    return String((value as { path: string }).path);
  return String(value ?? "");
}

// ─── Renderers (type-checked against schema definitions) ────────────

const demonstrationCatalogRenderers: CatalogRenderers<DemonstrationCatalogDefinitions> =
  {
    Title: ({ props }) => {
      const Tag = (
        props.level === "h1" ? "h1" : props.level === "h3" ? "h3" : "h2"
      ) as "h1" | "h2" | "h3";
      const sizes: Record<string, string> = {
        h1: "1.75rem",
        h2: "1.25rem",
        h3: "1rem",
      };
      return (
        <Tag
          style={{
            margin: 0,
            fontWeight: 600,
            fontSize: sizes[props.level ?? "h2"],
            color: "#111827",
            letterSpacing: "-0.01em",
          }}
        >
          {resolveText(props.text)}
        </Tag>
      );
    }

    Metric: ({ props }) => {
      const trendColors: Record<string, string> = {
        up: "#059669",
        down: "#dc2626",
        neutral: "#6b7280",
      };
      const trendIcons: Record<string, string> = {
        up: "↑",
        down: "↓",
        neutral: "→",
      };
      return (
        <div style={{ display: "flex", flexDirection: "column", gap: "4px" }}>
          <span
            style={{
              fontSize: "0.75rem",
              color: "#6b7280",
              fontWeight: 500,
              textTransform: "uppercase",
              letterSpacing: "0.05em",
            }}
          >
            {resolveText(props.label)}
          </span>
          <div style={{ display: "flex", alignItems: "baseline", gap: "8px" }}>
            <span
              style={{
                fontSize: "1.5rem",
                fontWeight: 700,
                color: "#111827",
                letterSpacing: "-0.02em",
              }}
            >
              {resolveText(props.value)}
            </span>
            {props.trend && props.trendValue && (
              <span
                style={{
                  fontSize: "0.8rem",
                  fontWeight: 500,
                  color: trendColors[props.trend] ?? "#6b7280",
                }}
              >
                {trendIcons[props.trend]} {resolveText(props.trendValue)}
              </span>
            )}
          </div>
        </div>
      );
    },

    PieChart: ({ props }) => {
      const COLORS = [
        "#3b82f6",
        "#8b5cf6",
        "#ec4899",
        "#f59e0b",
        "#10b981",
        "#6366f1",
      ];
      const data = props.data ?? [];
      return (
        <div style={{ width: "100%", height: 200 }}>
          <ResponsiveContainer>
            <RechartsPie>
              <Pie
                data={data}
                dataKey="value"
                nameKey="label"
                cx="50%"
                cy="50%"
                innerRadius={props.innerRadius ?? 40}
                outerRadius={80}
                paddingAngle={2}
              >
                {data.map((entry: any, i: number) => (
                  <Cell
                    key={i}
                    fill={entry.color ?? COLORS[i % COLORS.length]}
                  />
                ))}
              </Pie>
              <Tooltip />
            </RechartsPie>
          </ResponsiveContainer>
        </div>
      );
    },
    DataTable: ({ props }) => {
      const cols = props.columns ?? [];
      const rows = props.rows ?? [];
      return (
        <div style={{ overflowX: "auto", width: "100%" }}>
          <table
            style={{
              width: "100%",
              borderCollapse: "collapse",
              fontSize: "0.8rem",
            }}
          >
            <thead>
              <tr>
                {cols.map((col: any) => (
                  <th
                    key={col.key}
                    style={{
                      textAlign: "left",
                      padding: "8px 12px",
                      borderBottom: "2px solid #e5e7eb",
                      color: "#6b7280",
                      fontWeight: 600,
                      fontSize: "0.7rem",
                      textTransform: "uppercase",
                      letterSpacing: "0.05em",
                    }}
                  >
                    {col.label}
                  </th>
                ))}
              </tr>
            </thead>
            <tbody>
              {rows.map((row: any, i: number) => (
                <tr key={i} style={{ borderBottom: "1px solid #f3f4f6" }}>
                  {cols.map((col: any) => (
                    <td
                      key={col.key}
                      style={{ padding: "8px 12px", color: "#374151" }}
                    >
                      {String(row[col.key] ?? "")}
                    </td>
                  ))}
                </tr>
              ))}
            </tbody>
          </table>
        </div>
      );
    },
  };

// ─── Assembled Catalog ───────────────────────────────────────────────
// 封装定义和render逻辑 之后 注册到provider 上
export const demonstrationCatalog = createCatalog(
  demonstrationCatalogDefinitions,
  demonstrationCatalogRenderers,
  {
    catalogId: "copilotkit://app-dashboard-catalog",
    includeBasicCatalog: false,
  },
);

注册

import { StrictMode } from "react";
import { createRoot } from "react-dom/client";
import { CopilotKit } from "@copilotkit/react-core/v2";
import "@copilotkit/react-core/v2/styles.css";
import { demonstrationCatalog } from "./catalog/renderers";
import "./globals.css";
import App from "./App";

createRoot(document.getElementById("root")!).render(
  <StrictMode>
    <main className="h-screen w-screen">
      <CopilotKit
        useSingleEndpoint={false}
        runtimeUrl="/api/copilotkit"

        // 🪁 Enable A2UI using the demonstration catalog
        a2ui={{ catalog: demonstrationCatalog }}
      >
        <App />
      </CopilotKit>
    </main>
  </StrictMode>,
);

fixed schema UI

固定模式（Fixed Schema）声明式生成式 UI

使用固定模式时，你需要提前设计好 A2UI 组件树——布局、嵌套结构和数据绑定都已预先定义。Agent 的唯一工作就是填充运行时数据。这让你可以最大程度地控制最终 UI，同时仍由 Agent 来决定何时展示以及展示哪些数据。

当你需要为特定场景（例如航班卡片轮播）保持一致、精致的布局，且不需要 Agent 即兴发挥结构时，这种方式非常适用。

构建固定模式最简单的方式是使用 https://a2ui-editor.ag-ui.com/ 编辑器。打开 Composer（组合器）并指定以下提示词。

将ui schema 注册到后端agent 上

from copilotkit import a2ui
from langchain.tools import tool

CATALOG_ID = "copilotkit://app-dashboard-catalog"
SURFACE_ID = "flight-search-results"

FLIGHT_SCHEMA = [....] // 定义航班卡片的schema

`@tool
def display_flights(flights: list[Flight]) -> str:
    """Display flights as rich cards in a horizontal row.

    Each flight must have: id, airline, airlineLogo (URL), flightNumber,
    origin, destination, date, departureTime, arrivalTime, duration,
    status, and price.
    """
    return a2ui.render(
        operations=[
            a2ui.create_surface(SURFACE_ID, catalog_id=CATALOG_ID),
            a2ui.update_components(SURFACE_ID, FLIGHT_SCHEMA), // 按schema 渲染
            a2ui.update_data_model(SURFACE_ID, {"flights": flights}),
        ],
    )

graph = create_agent(
    model=ChatOpenAI(model="gpt-4.1"),
    tools=[get_sales_data, search_flights, display_flights],
    middleware=[CopilotKitMiddleware()],
    checkpointer=MemorySaver(),
    system_prompt=(
        "....."
    ),
)

Dynamic vs Fixed: When to use which

	Fixed Schema	Dynamic Schema
Layout	Predefined, identical every time 预先定义好的，每次都是相同的	Agent-generated, varies per request由代理生成，随每个请求而不同
Agent’s role	Fills in data only 仅填充数据	Chooses components and layout 选择了组件和布局方式
Consistency 一致性	Maximum最大值	Varies 各不相同
Flexibility 灵活性	Minimal — new layouts require code changes 非常少——新的布局需要修改代码	High — agent adapts to the request 高——代理人能够适应该请求
Best for	Polished, known surfaces (flight cards, invoices) 经过精心处理的、易于识别的表面（如飞行卡片、发票）	Long-tail, exploratory, or internal surfaces 长尾的、探索性的或内部表面

在实践中，许多应用程序同时使用两种方案：对于高流量且需要注重品牌形象的页面，使用固定的模式；而对于其他所有页面，则使用动态的模式。

声明式生成式用户界面允许智能体从一系列构建模块中组合出各种界面——这种方式比受控式界面更加灵活，也比自由式界面更加稳定。

A2UI 规范包含了三个核心要素：组件目录（包含组件的定义和渲染方式）、架构设计（组件之间的布局方式），以及数据绑定（运行时的值的传递方式）。

动态模式允许代理实时生成布局——这对于需要长尾布局和探索性界面的情况非常有用。

固定式的架构提供了预定义的布局，代理程序可以将数据填充到这些布局中——这种方式适用于需要呈现精美外观且流量较大的场景。

这两种方法可以在同一个代理中同时存在。

Open Generative UI 开放式

开放式的生成 UI 是最灵活的模式：该工具并不局限于一组预先注册好的组件，也不依赖于固定的或动态生成的声明式架构。

这种灵活性源自 MCP 应用程序。代理可以从 MCP 服务器上获取应用程序工具，并在这些工具符合用户需求时将其打开使用。

你的前端部分充当了主机角色。它无需提前手动编写所有可能的用户界面界面；相反，它只需将聊天运行时与兼容的外部应用程序连接起来即可。

为什么使用开放式用户界面
当用户的请求范围过于广泛，无法用固定的组件库来涵盖时，开放式用户界面就非常有用。
优点

极其灵活：该代理能够引导用户体验更丰富的应用程序功能，而不仅仅是简单的内嵌小部件。
降低前端耦合度：通过连接到 MCP 服务器，宿主应用程序可以获得新的功能。
非常适合用于白板演示、设计工作、计划制定等需要使用工具来完成的任务。
缺点
与那些需要手动配置或声明式的方法相比，这种方式对最终的用户界面控制较少。
质量取决于所连接的 MCP 应用程序，以及代理程序选择这些应用程序的精准程度。
这需要更强的信任机制、权限管理，以及更严格的集成规范。

MCP 应用程序规范
MCP 应用程序是模型上下文协议的扩展，它使得 MCP 服务器能够向支持的设备提供交互式用户界面。

这种架构由三部分组成：

服务器：提供工具和用户界面资源。
HOST：将用户界面嵌入到沙盒中的 iframe 中，并负责代理双方的通信过程
视图：在 iframe 内部运行的应用程序。
一个重要的设计原则是渐进式增强：如果host支持 MCP 应用程序，那么工具会呈现丰富的用户界面；否则，它仍然可以作为普通的 MCP 工具使用，仅提供文本输出。

CopilotKit 负责托管工作——你只需要将其指向一个 MCP 服务器的 URL 地址即可。

the official repository of MCP App examples.

注意：开放生成的用户界面是不可预测的

因为代理在每次请求时都会从零开始生成用户界面，所以结果可能不会完全符合你的预期。你可能需要多次调整提示语，直到得到你想要的效果。

这是完全开放式的生成式用户界面所面临的核心权衡之一：虽然可以获得最大的灵活性，但必须牺牲一致性和可预测性。对于那些需要始终保持可靠性的用户界面来说，采用控制性或声明式的方法更为合适。

import { LangGraphHttpAgent } from "@copilotkit/runtime/langgraph";
import {
  CopilotRuntime,
  createCopilotEndpoint,
} from "@copilotkit/runtime/v2";
import { serve } from "@hono/node-server";

const appAgent = new LangGraphHttpAgent({
  url: process.env.LANGGRAPH_DEPLOYMENT_URL || "http://localhost:8005",
});

/***
 * 
 * 关于 mcpApps 配置的几点说明：
 * mcpApps.servers 将运行时连接到一个或多个提供应用程序工具的 MCP 服务器。
 * 每个条目都指明了运行时应该在哪里找到兼容的 MCP 应用程序。在这里，你可以通过 HTTP 连接到 Excalidraw。
 * CopilotKit 会自动为代理添加 MCP 应用程序识别功能，从而在运行时能够自动调用这些工具。
 */
const runtime = new CopilotRuntime({
  agents: { default: appAgent },
  openGenerativeUI: true, // 启用开放生成式UI 代理可以生成任意类型的用户界面——包括 HTML、CSS、JavaScript 等代码，并可以直接在聊天界面中展示
  mcpApps: { // 注册 MCP 应用程序
    servers: [
      {
        type: "http",
        url: "https://mcp.excalidraw.com", // <- Exalidraw MCP Server
        serverId: "example_mcp_server",
      },
    ],
  },
});

const app = createCopilotEndpoint({
  runtime,
  basePath: "/api/copilotkit",
});

serve({ fetch: app.fetch, port: 4005 }, () => {
  console.log("✓ CopilotKit API server running at http://localhost:4005");
});

需要注意，全开放式的生成逻辑不同于以上两种，会自动用iframe 包裹要展示的内容，无论是借助mcpApps 还是开启openGenerativeUI，都会自动用iframe 包裹要展示的内容。

全栈状态共享

在后端agent 服务中，使用AgentState 声明状态值，和更新tool

from __future__ import annotations

import uuid

from langchain.agents.middleware import AgentState as BaseAgentState
from langchain.tools import ToolRuntime, tool
from langchain_core.messages import ToolMessage
from langgraph.types import Command
from typing_extensions import TypedDict


class Todo(TypedDict):
    id: str
    title: str
    completed: bool


class AgentState(BaseAgentState):
    todos: list[Todo]

@tool
def manage_todos(todos: list[Todo], runtime: ToolRuntime) -> Command:
    """Replace the entire todo list. Use this to add, edit, or remove todos."""
    for todo in todos:
        if not todo.get("id"):
            todo["id"] = str(uuid.uuid4())

    # Command allows agents to update state via toolcalls
    return Command(update={
        "todos": todos,
        "messages": [
            ToolMessage(
                content="Successfully updated todos",
                tool_call_id=runtime.tool_call_id,
            )
        ],
    })


@tool
def get_todos(runtime: ToolRuntime):
    """Get the current todo list."""
    return runtime.state.get("todos", [])


todo_tools = [manage_todos, get_todos]

graph = create_agent(
    model=ChatOpenAI(
        openai_api_key=ARK_API_KEY,
        openai_api_base=ARK_API_BASE,
        model_name=ARK_MODEL_NAME, # 从环境变量读取模型
    ),
    state_schema=AgentState,
    tools=todo_tools,
    middleware=[CopilotKitMiddleware()],
    checkpointer=MemorySaver(),
    system_prompt=(
        "You manage a shared todo list. "
        "Use manage_todos to add, edit, or remove todos. "
        "Use get_todos to check the current list. "
        "When asked to manage todos, call the openOrCloseTodos frontend tool with open=true first. "
        "Keep responses to 1-2 sentences."
    ),
)

使用useAgent 可以实时在前端获取后端agent中注册的状态数据
使用useFrontendTool 可以给后端agent 注册tool 进行页面操作

import { useState } from "react";
import { z } from "zod"

import { TodoAppLayout } from "@/components/todo-app-layout";
import { TodoList } from "@/components/todo-list";
import { CopilotChat, useAgent, useFrontendTool } from "@copilotkit/react-core/v2";

/**
 * useAgent() gives you a live handle to the agent's shared state: agent.state.todos reflects the current list, and agent.setState({ todos: updated }) pushes changes back.
 * useFrontendTool registers a tool that runs in the browser — the agent calls it like any backend tool, but the handler executes client-side and can update React state directly.
 */
export default function App() {
  const [todosOpen, setTodosOpen] = useState(false);

  // 🪁 Register a frontend tool the agent can call to control the UI
  useFrontendTool({
    name: "openOrCloseTodos",
    description: "Open or close the todo panel.",
    parameters: z.object({ open: z.boolean()}),
    handler: async ({open}) => {
      setTodosOpen(open);
      return `Todos are ${ open ? 'open' : 'closed'}.`;
    },
  });

  // 🪁 Subscribe to shared agent state
  const { agent } = useAgent();

  return (
    <TodoAppLayout
      chat={<CopilotChat />}
      open={todosOpen}
      onOpenChange={setTodosOpen}
      panel={(onClose) => (
        <TodoList
          // 🪁 Read shared state
          todos={agent.state.todos || []} 

          // 🪁 Write shared state
          onUpdate={(updated) => agent.setState({ todos: updated })}

          isRunning={agent.isRunning}
          onClose={onClose}
        />
      )}
    />
  );
}

Sequence to Sequence Models

发表于 2026-07-17

Sequence to Sequence Models

输入为一个序列，输出为另一个序列

比如语言翻译

根据图片内容生成对其语言描述

机器翻译过程可以看做是一个有条件的语言模型

找到最合适的翻译结果Y 的过程，就是计算在 x 输入情况下得到Y的概率P，使P达到最大值的过程

相关练习
上述过程更适合直接获取到整个句子Y后，对整个句子的适配度进行概率计算，而不是用贪心算法，
依次根据上一个词的内容找到最匹配的下一个词之后，再最终对整个句子进行概率运算，
因为单个词语查找最合适下一个词语的概率，是计算量非常大的，另外词语的高频性可能会影响句子整体的适配性
比如下面句1 比句2 翻译的效果更好
但是句1和句2 在用贪心算法获取过程中更可能得到句2，因为得到相同的前两个词语之后，going 出现的频率大于visiting

贪心算法的缺点是，在获取到当前词语之后，会根据当前词语，获取下一个词语，然后获取下一个词语，直到获取到整个句子.

集束搜索算法

集束搜索算法（beam search algorithm）: 在获取当前词语之后，会获取当前词语的概率最大的k个词语，然后获取下一个词语，直到获取到整个句子
相当于同时进行集束宽度（eg. B=3）个贪心算法，然后再每一步的n*3 个结果中寻找概率最大前B 个结果，然后进行下一步.n 是当前词汇库中的词汇数
这样就比贪心算法可能得到更多的可能性

优化

当前的概率计算存在以下问题

概率值都是小于1的值，多项相乘时，项数越多，值越小，对于计算来说，能存储的极小值有限，容易出现数据下溢
基于问题1，项越多概率越小，导致模型为使概率值最大，尽可能生成短句子，从而天然倾向生成较短的句子，不倾向生成长句子
对于长句生成本身不友好

对于集束宽度的选择
实际生产为10个左右，大于100的基本就可以算做太多了
large B 可以得到更好的结果，但是训练时间会变长
small B 会得到差的结果，但是训练时间会变短

虽然集束搜索算法不如深度或者广度优先搜索算法的准确性高，不能保证找到最准确的值，但是计算速度会比二者快

问题分析

通常增加测试数据集或者增加集束搜索宽度，一般都会是效果更好
但判别是RNN过程有错误，还是集束搜索算法搜索过程有错误可以更根本的解决训练异常问题

如果人工翻译结果概率大于算法得出的句子的概率，即集束算法没有找到使概率最大的正确的句子，问题出在解码查找过程
相反，在本身人工翻译结果好于算法结果的情况下，如果人工翻译结果概率还是小于算法得出的句子的概率，说明RNN过程有错误

bleu 算法

类似图像识别可以用对错一个确定的数值判断推测结果，这里用bleu 算法得到的指数来评估翻译结果好坏
用来计算评估机器翻译结果的指数，从而评估翻译好坏
翻译越接近人类翻译，指数分数越高
具备双语评估，且具有人类评估结果替补性

短翻译容易获得高精度的评判结果，为了降低这种误差，引入BP 惩罚参数进行更精准效果计算

注意力机制

之前涉及的机器翻译是使用两个rnn 结构，先编码后解码（编码后将信息存在激活函数中），一次性读取整个句子会之后再整体翻译
这样对于特别长的句子来说不太现实，即需要处理的入参特别长，随着句子长度增加，bleu 指数会有明显下降
且不符合人类翻译习惯，人类翻译更倾向于，读一段翻译一段，这样只关注一整段文字里面的一小段句子

注意力机制在编码阶段会生成一系列输入单词的特征信息（单词本身和它周围词的特征），
在生成翻译结果时，根据输入句子中每个单词对该翻译结果单词的影响权重（注意力权重）和已经翻译出的单词两个信息共同计算出当前单词的翻译结果

具体计算过程

更多应用，日期格式转换

应用

语音识别，输入变成每一帧的语音音频, 1秒有1000帧的话，输入就会是1000个数据，相当于上面1000个单词

触发词识别

transformer

对于循环神经网络来说，rnn, Lstm 处理连续输入信息会要求神经网络越来越复杂

将注意力机制中RNN 过程用CNN 卷积同时处理多个单词的过程就是transformer 架构

transformer 架构的模型两个关键点

自注意力机制 — 带着句子上下文情况下特征计算
多头注意机制 — 自注意力机制循环过程，产生多种可能性

自注意力机制

计算输入单词的注意力值

多头注意力机制

在自注意力机制基础上，同时并计算其他多个问题的注意力值
通过并行同时获取多个单词之间的不同维度的含义关系

transform 网络

有时候使用mask multi-head attention 进行训练
在有准确输入输出情况下，
在解码阶段，将输入值从正确输出结果里一次掩盖后几个单词，假设还没有推理出来，仅推理出了前几个单词
查看能否正确预测出序列中的下一个单词

Embedding 词嵌入

发表于 2026-07-16

word representation 词汇表征

如果用one-hot表示词汇集中的word，只能存储这个单词在这个词表中的索引，无法表示单词的语义，更无法计算不同单词之间的相似度/关联度。
如果选取n个features, 每个feature 再用数据大小表征当前单词和这个feature的关联度，可以更好的表征单词语义和计算单词之间的相似度，从而进行词嵌入，进一步进行猜词计算

引入坐标系的概念，如果n个特征代表n 个维度,比如300个维度，那么每个单词就像嵌入在这个300维空间中, 会自然的发现类似的或者同类的单词会聚集在一起
使用t-sne 算法通过降维到2维可以实现词分布的可视化

词嵌入应用

实体识别 ==> 迁移学习

因为现在每个单词是通过词嵌入表征的，所以如果对于已经识别的实体，对于现在要识别的实体，只需要对比特征向量的相似度，就可以进行识别
另外，对于词汇表中未出现的单词，通过进行特征向量转换后，自然可以加入词汇表中，从而可以直接进行相似度计算

词嵌入方便进行迁移学习，尤其是大批量词嵌入模型应用与小批量垂类领域时，需要重新进行词嵌入的新词是少数，可以直接使用词嵌入模型中已存在的大多数语义逻辑，从而进行迁移学习

词嵌入迁移过程

词嵌入word embedding 和之前的人脸识别的 face encoding 思路一致，可以认为是意思一样的术语

更多：文本摘要（text summarization）文本解析（co-reference resolution）解析（ parsing）

类比推理

使用余弦相似度算法或者其他可以计算相似度的算法，可以对两个已知单词进行向量计算，得到向量关系，从而推理出问题中单词在相同向量，即类比关系中对应的单词

比如 Q: man 对应 woman 那么 king 对应 ?
a: eman - e_woman = e_king - e?

e? = e_king - e_man + e_woman
如果向量 e? 和 e_king - e_man + e_woman 相似度值越高，即词语语义越相似，越接近正确答案

使用余弦计算相似度
两个向量夹角角度的余弦值，夹角为90度时，余弦值为0, 夹角为0度时，余弦值为1，夹角为180度时，余弦值为-1，相反方向

词嵌入矩阵

所有词汇表中的单词进行特征向量表示后，组成的矩阵，如果跟原来one-shot表示法中某个单词w的向量进行点积运算可以直接得到该单词w 的特征向量表示
E · O_w = e_w

使用word embedding 进行模型训练

整体步骤都是将样本数据（一个句子）中单词分为目标值和输入值后，形成监督学习模型，然后进行训练

将目标值转成特征向量，摊平所有的向量值
引入softmax 激活函数，计算目标值出现的概率，取概率最大的值与目标值对比
计算损失函数，重新训练

词嵌入矩阵的训练过程

优化

固定输入单词数量，如上例，输入值为推测词前面的所有6个单词，300维的特征摊平后，输入值为300 * 6 = 1800维
如果固定单词数量为4，输入单词数量减少，那么特征向量的维度也会减少，模型输入值减少，从而可以减少训练时间
固定输入单词数着窗口，可以方便的处理任意长度的句子，因为这样模型的输入结构就是固定的
尝试使用不同的上下文选取窗口大小进行训练
最近4个
左右各4个
最近1个
附件的一个
使用 Word2Vec 中的 Skip-grams 算法进行训练
Skip-grams 先固定输入上下文，在根据不同窗口规则选取目标值，生成监督学习数据对

softmax 分类的问题就是要全量计算所有的概率，然后取概率最大的值，这样计算量非常大

负采样
选取一对正确的目标值和上下文，作为输入，输出为正确的flag 1，作为正采样样本
然后随机生成k个负采样样本，负采样样本输入值中的上下文和正采样样本一致，目标值随机选取，输出为错误的flage 0

正采样样本和负采样样本一起组成监督学习样本对进行训练
这样做的好处，相当于原来要判断单词集合中所有单词是不是正确答案，错误答案是远多于正确答案的，现在只要判断k个单词生成的错误答案，避免了每次判断所有的单词，从而大大减少计算量

负采样单词的获取为了避免取到极高频词或者极低频词，一般用下面的概率公式判断当前单词概率，决定是否要采样，f(w) 为单词出现频率
情感分类
将评论中单词进行特征向量转换后进行均值计算，最后对计算结果进行softmax 分类

但这样没有考虑单词之间的语序和重复出现的频率问题，会导致预测结果不准确
可以使用RNN 进行训练, 解决单词语序引起的语义失真问题

相关练习
语义纠偏

一些实操的步骤

发表于 2025-11-01

文生视频

找到相似图片，通过文心一言生成图片提示词
将提示词替换下面代码, 在jupyter notebook中运行可以得到5张图片

import json
import os
import dashscope
from dashscope import MultiModalConversation

# 以下为北京地域url，若使用新加坡地域的模型，需将url替换为：https://dashscope-intl.aliyuncs.com/api/v1
dashscope.base_http_api_url = 'https://dashscope.aliyuncs.com/api/v1'

messages = [
    {
        "role": "user",
        "content": [
            {"text": "一张充满秋日氛围的摄影图，画面中心是一片金黄的枫叶，叶脉清晰、边缘略带枯黄斑点，阳光从画面右侧洒下，形成耀眼的光斑与柔和的光线；背景是暖色调的树木（树叶呈橙红渐变），远处有朦胧的建筑轮廓；整体风格自然写实，光线明亮且带暖调，营造出宁静又诗意的秋日氛围，细节上突出枫叶的质感与光影层次。"}
        ]
    }
]

# 若没有配置环境变量，请用百炼API Key将下行替换为：api_key="sk-xxx"
# 新加坡和北京地域的API Key不同。获取API Key：https://help.aliyun.com/zh/model-studio/get-api-key
api_key = "xxxxx"
for  i in range(0,5):
    response = MultiModalConversation.call(
        api_key=api_key,
        model="qwen-image-plus",
        messages=messages,
        result_format='message',
        stream=False,
        watermark=True,
        prompt_extend=True,
        negative_prompt='',
        size='1328*1328'
    )
    
    if response.status_code == 200:
        print(json.dumps(response, ensure_ascii=False))
        import requests
        url = json.dumps(response, ensure_ascii=False)
        reponse = requests.get(url=response.output.choices[0].message.content[0]['image'])
        with open('./ad50000'+str(i)+'.png','wb') as f:
            f.write(reponse.content)
    else:
        print(f"HTTP返回码：{response.status_code}")
        print(f"错误码：{response.code}")
        print(f"错误信息：{response.message}")
        print("请参考文档：https://help.aliyun.com/zh/model-studio/developer-reference/error-code")

将图片上传到度加-视频生成，生成依据图片生成的视频

商品评论分析

# Python3.8以上
pip install pydantic==2.5.3 -i https://mirrors.aliyun.com/pypi/simple/
pip install  langchain-core==0.1.45 -i https://mirrors.aliyun.com/pypi/simple/
pip install langchain==0.1.16 -i https://mirrors.aliyun.com/pypi/simple/
pip install  dashscope==1.24.7 -i https://mirrors.aliyun.com/pypi/simple/

#导入库
import os
import pandas as pd
from typing import List, Optional
from langchain_core.pydantic_v1 import BaseModel, Field
from langchain_core.utils.function_calling import convert_to_openai_tool
from dashscope import Generation
import dashscope
from langchain.tools.render import render_text_description
import warnings
warnings.filterwarnings('ignore')

# 1.初始化大模型环境

# 使用大模型类型
os.environ["DASHSCOPE_API_KEY"] = "https://dashscope.aliyuncs.com/compatible-mode/v1"

dashscope.api_key = "xxx" // 阿里云百炼 https://bailian.console.aliyun.com/?tab=model#/api-key

from typing import Optional

from langchain_core.pydantic_v1 import BaseModel, Field


class Person(BaseModel):
    """Information about a person."""
    name: Optional[str] = Field(default=None, description="The name of the person")
    hair_color: Optional[str] = Field(default=None, description="The color of the peron's hair if known")
    height_in_meters: Optional[str] = Field(default=None, description="Height measured in meters")

2.读取数据

df=pd.read_excel(r"./原始数据.xlsx")

df.head(2)
df[df["Rating"]<2].groupby("Clothing_ID").size().sort_values().tail(10) # 评分低于2的最多的10条数据

3.筛选数据
re_862=df[(df["Rating"]<2)&(df["Clothing_ID"]==862)].Review_Text.dropna(axis=0,how=any).tolist() // 去掉Review_Text 缺省的数据

4.大模型处理

# 定义问题类别
class clothing(BaseModel):
    Fabric_problems: Optional[str] = Field(default=None, description="Fabric problems: noticeable holes, loose threads, pilling, or abnormal color fading and shrinkage")
    Workmanship_defects: Optional[str] = Field(default=None, description="Workmanship defects: poor stitching, excessive loose threads, easily detachable buttons, etc")
    Size_discrepancies : Optional[str] = Field(default=None, description="Size discrepancies: the actual product size differs from the described dimensions, leading to ill-fitting garments.")
    Uncomfortable_wear : Optional[str] = Field(default=None, description="non-breathable fabrics, irritation against the skin, or discomfort during wear.")
    Fit_design: Optional[str] = Field(default=None, description="Fit and design: unsuitable cut, not conforming to ergonomics, or unflattering when worn")
    Appearance_Discrepancies: Optional[str] = Field(default=None, description="Mismatch between the product received and online images in terms of style or color.")
    After_Sales: Optional[str] = Field(default=None, description="Unsatisfactory customer service attitude or response efficiency.")
    Value_Money: Optional[str] = Field(default=None, description="Consumers feel that the price does not align with the actual quality or value of the product, perceiving it as overpriced.")

class Data_yun(BaseModel):
    """提取负面信息"""
    comment: List[clothing]

# 封装请求数据
def prompt_yun(ct):
    system_prompt="您是提取算法专家。仅从文本中提取相关信息。如果您不知道要求提取的属性值，则为属性值返回 null。首先抽取出用户评价中的抱怨点，然后对这些点进行分类，然后调用工具输出"
    prompt=[{"content":system_prompt,"role":"system"},\
            {"role":"user","content":"The material is cheap"},\
            {'role': 'assistant', 'content': '', 'tool_calls_ids': [{'function': {'name': 'Data_yun', 'arguments': 
                                                                              '{"comment": [{"Fabric_problems": "The material is cheap","Size_discrepancies": "is made for women with long torso and shorter shoulders"}]}'}}]}]
    prompt.append({"role":"user","content":ct})
    return prompt

# 向大模型提问
def get_response_1t(mess):
    response = Generation.call(
        model='qwen-plus',
        messages=mess,
        tools=[convert_to_openai_tool(Data_yun)],
        result_format='message', # 将输出设置为message形式
    )
    return response

5.处理单个评论，进行模型能力测试

text = re_862[1]
p_y = prompt_yun(text)
res = get_response_1t(p_y)

res.output.choices[0].message["tool_calls"][0]['function']['arguments']


6.循环处理所有数据

import time
import numpy as np
for i in re_862:
    time.sleep(0)
    text = i
    if text is np.nan:
        print("数据为空值")
    else:
        p_y=prompt_yun(text)
        res=get_response_1t(p_y)
        print(res.output.choices[0].message["tool_calls"][0]['function']['arguments'])
        print('-'*20)

进行模型训练的其他方式

发表于 2025-10-25

csv 文本文件，没有多余格式，适用于各种场景使用，不仅适用各种算法，也适用数仓

jupyter notebook
执行: shift + 回车
注释：control + / ===>解释性文字
收起执行的结果：点击蓝色边框

python install pandas numpy
pandas numpy 数据分析，数据挖掘

建模

拆分数据：

纵向拆分
x: 除目标列以外的列
y: 目标列

横向拆分：
训练集：模型学习
测试集：验证模型效果

特征处理

特征缩放
数值类型的数据取值范围差异较大，导致模型在建模的时候忽略了某些属性对模型的重要度，
所以就需要通过特征缩放来平衡属性之间的权重

方法：标准化，最小值，最大值归一化

特征离散化
离散化：将连续数据转换成离散数据
数值的属性范围较大导致模型进行判断条件变多，进而影响模型效率
方法：
分箱： 1,1,1,1,1,1,4,5,1,10
等宽分箱：每个区间的范围大小是相同的
0-5 6-10
[1,1,1,1,1,1,4,5] [10]
等频分箱：每个区间的样本数量相同
[1,1,1,1,1] [1,1,4,5,10]

基于聚类的离散化
基于信息熵的；离散化
基于卡方的离散化
1R

特征编码
有一些模型无法处理字符串类型的数据，这个时候就需要通过特征编码将字符串类型数据转换成模型可以处理的数据

方法：

label-encoding(标签编码)：将字符串映射为对应数字eg: male –> 0, female –> 1，应用于二元属性和序数属性
one-hot-encoding(独热编码)：将字符串转换为不同的01组合，应用于标称属性（没有大小顺利的特征）

代码的方式

读取文件

pip install pandas
pip install numpy


df = pd.read_csv('./wine3.csv')
df.head()

缺失值处理方式

删除：

按行删除：

按列删除：该列数据超过20%缺失，可以考虑删除该列

填充：通过特殊值填充，如：均值（数据为浮点性），中位数，众数（数据字符串），0

sklearn 也可以处理缺失值，填充算法预测


df.isnull().sum()

df.shape # 查看行列数

del df['phenols'] # 删除列

df.shape

df.info() # 查看每列数据类型

<!-- np.nan 代表缺失值，可能是上面提到的三种 -->

# 填充
df['malic'] = df['malic'].replace(np.nan,df['malic'].mean()) 

# aixs : 0 按行删除； 
# how :any 只要有一个数据缺失就将该行删除，all 表示这一行全部缺失的时候才删除

df_drop = df.dropna(axis= 0, how='any') 

df_drop.isnull().sum()

异常值处理

异常值：

偏离正常样本过多的值，也叫噪声点，离群点，一般是极大或者极小的值

但不是错误值，只有数值类型（有数值含义）数据才会有异常值

处理：同缺失值
删除：按行删除
替换：使用特殊值替换，（当做缺失值处理）

异常值检测

箱线图
Q3 + 1.5IQR < X < Q1 - 1.5IQR
IQR = Q3 - Q1
Q3: 75% 分位数 Q: 25% 分位数
1.5 : 经验值，可容忍范围，可根据实际情况调整

3σ 原则: z-score
|x- μ| > 3σ
σ 标准差 μ 均值

务必根据实际情况分析，比如年龄,可以设定 1-100 范围为正常值，则不需要再进行上述检测

1
2
3

import matplotlib.pyplot as plt
plt.boxplot(df_drop['malic'])
plt.show()

3.1 箱线图：进行异常值检测

while True:
    list_error = []
    Q1 = df_drop['malic'].quantile(0.25)
    Q3 = df_drop['malic'].quantile(0.75)
    IQR = Q3 - Q1
    up = Q3 + 1.5*IQR
    down = Q1 - 1.5*IQR
    n = 0
    for x in df_drop['malic']:
        if x > up or x<down :
            print(x)
            df_drop.loc[:, 'malic'] = df_drop['malic'].replace(x,df_drop['malic'].mean()) # 填充 
            list_error.append(x)
    if len(list_error) == 0:
        break


for x in df_drop['malic']:
    if x > up or x<down :
        print(x)
plt.boxplot(df_drop['malic'])
plt.show()

3.2 3σ 原则: z-score
|x- μ| > 3σ


mport  warnings
warnings.filterwarnings('ignore')

mean = df_drop['ash'].mean()
std = np.std(df_drop['ash'])
for x in df_drop['ash'].index:
    if np.abs(df_drop['ash'][x] - mean)>3*std:
        print(df_drop['ash'][x])
        df_drop.drop(axis=0,index=x,inplace=True)

拆分数据

# 纵向拆分
X= df_drop.drop(axis=1, columns='type')
Y = df_drop[['type']]

pip install scikit-learn

from sklearn.model_selection import train_test_split

X_train,X_test,Y_train,Y_test = train_test_split(X,Y,test_size=0.3)

建模

from sklearn.neighbors import KNeighborsClassifier

model_knn = KNeighborsClassifier()
model_knn.fit(X_train, Y_train)

评估

1	model_knn.score(X_test,Y_test)

保存模型

pip install joblib

import joblib

joblib.dump(model_knn, './红酒等级预测.pkl')

8 预测

1
2
3


import pandas as pd
import joblib

8.1 读取数据

1 2	df = pd.read_csv('./wine3_pre.csv') df.head()

8.2. 读取模型

1	model = joblib.load('./红酒等级预测.pkl')

8.3. 预测数据

<!-- 把要预测的数据处理成跟训练的时候的维度一样 -->
del df['phenols']
model.predict(df)

Y_pre = model.predict(df) # 预测
df['type'] = Y_pre # 合并数据

8.4 存储预测数据

1	df.to_csv('./wine_model_pre.csv', index=False)

在线练习

使用 knime

软件下载，要翻墙 https://www.knime.com/downloads

创建工作流

添加数据源
将csv Reader 拖到工作流绘制中心，并只指定训练数据文件

切换 transformation tab 可以指定要保留的列数据，和修改该列数据类型

也可以再后续添加 string Replacer 进行具体数值转换

看起来一次只能改一个值，多个值进行转换，可以添加多个string Replacer

缺失异常值处理

添加misssing Value, 对常见数据类型添加缺失值处理规则

也可以对指定列进行缺失值处理规则指定

训练模型
选择算法指定，目标列
注意knime 只允许数据类型为string 的列作为目标列，不是的话，要在前面步骤中进行转换
模型预测和结果保存
注意训练模型的算法和预测的算法要一致
添加模型预测，结果保存,csv Writer 需要指定保存的文件路径名称

本地运行大模型的一些选择 & 指令和图形化微调

发表于 2025-09-27

业务场景分类

文本处理：LLM（自然语言处理）

视频图片：计算机视觉

混合模型：文本、图片、视频、音频。

任务类型

训练任务：通常是数据量较大，导致计算量也比较大。并行度也高。对资源消耗比较大。对时效性没有太大要求。
推理任务：是使用已有大模型对现有情况处理，数据量通常不大，所以资源消耗没那么大。通常要求处理速度快。

硬件选择

服务器：要评估业务的资源使用情况来具体配置
国产化GPU：昇腾
国产化CPU：华为、海光
操作系统：Linux、欧拉、麒麟

还要考虑以上4者之间的兼容性。

计算资源

全部使用GPU：耗费高，适用于中大型企业，
GPU+CPU的组合：适用于中小企业部署，如果用CPU 替换GPU, 所用CPU 数量将是原理GPU 数量的两倍。

华为云 - 工作台 - AI NOTEBOOK 支持使用NPU 进行训练

阿里云 - MODEScope - My Notebook 支持使用GPU

模型选择/模型类型

L0：通用场景的模型，但是通常精度不高

L1：进行微调以后的模型，可以适用某一个特定的领域

微调方式：代码、命令行、图形化界面

L2：只专注某个功能的模型，如：代码生成

借助LLAMAFactory 进行微调

LLAMAFactory

git clone https://github.com/hiyouga/LLaMA-Factory.git

cd LLaMA-Factory

pip install -e ".[torch,metrics]" 

# 如遇版本冲突可尝试用下面的命令

pip install --no-deps -e .
# 
llamafactory-cli version

# 方式一： 启动webui，通过页面配置进行微调
llamafactory-cli webui

# 方式二： 命令行进行微调 llamafactory-cli train 配置文件路径，文件和界面作用一样，只不过用key=value进行配置

llamafactory-cli train examples/train_lora/llama3_lora_sft.yaml 

# 安装nvitop 查看内存使用情况

pip install nvitop
nvitop -m auto

指定数据集

配置文件：data/dataset_info.json

数据集都放在data文件夹下面，通过配置dataset_info.json文件，可以更新界面上数据集选项
也可以根据界面上数据集选项生成规则对命令行调用的文件进行配置

训练后大模型产物一般会存放在save文件夹中，或者根据配置的输出路径进行存放

用界面调试

1 2	# 方式一：启动webui，通过页面配置进行微调 llamafactory-cli webui

模型下载源可指定modelScope，或者HuggingFace 里面的模型,避免下载到本地

用命令行调试

1
2
3

# 方式二： 命令行进行微调 llamafactory-cli train 配置文件路径，文件和界面作用一样，只不过用key=value进行配置

llamafactory-cli train examples/train_lora/llama3_lora_sft.yaml

通过修改配置文件达到使用界面一样的配置效果

nvitop 查看内存

调试完成

不管通过命令行调试还是通过界面调试，都可以通过查看日志判断是否完成调试或者拍错排查

模型保存、模型导出

通过界面保存

注意看保存到文件下面的几个关键文件有没有

保存后需要可用ollama 命令行查看是否正常导出

1	ollama show --modelfile Qwen2.5:0.5B

通过命令行保存

修改examples/merge_lora 文件夹下对应文件，指定模型现存位置和导出位置

// 运行命令导出
llamafactory-cli export examples/merge_lora/llama3_lora_sft.yaml

// 查看模型相关信息，进一步确保模型正确导出

ollama show --modelfile Qwen2.5:0.5B

// 安装压缩命令行，对模型进行压缩
sudo add-apt-repository multiverse   
sudo apt update
sudo apt installsudo apt install unrar rar

// 把导出的模型进行压缩
rar a -r Qwen3-8rar a -r Qwen3-8B_lora_sft.rar ./output/Qwen3-8B_lora_sft/                                                        se

模型使用

通过界面的形式模型在本地不用下载
通过命令行模型在本地使用要先下载到本地
下载后进行解压

1
2
3

// 切换到导出模型的文件夹执行[部署]命令
ollama create qwen2.5-wt0.5b -f Modelfile
// 出现”success”表示模型部署成功。

部署模型后可以就可以通过open-webui 进行使用了

1	open-webui server

参考

微调训练：包括以下几部分配置：
微调训练的阶段：预训练、指令微调、强化学习等等；
微调使用的数据集：数据集的格式、路径、验证集的比例等等；
关键微调参数：在我们之前教程中重点学习的：学习率、训练轮数、批量大小等等；
LoRA 参数：当使用 Lora 微调时需要配置的一些特殊参数；
RLHF 参数：当训练阶段为强化学习时需要配置的一些特殊参数；
特殊优化参数：当选择使用 GaLore、APOLLO 和 BAdam 优化器时需要配置的一些参数；
SwanLab 参数：一款开源的模型训练跟踪与可视化工具；

LLM模型选择
成本
本身训练语言是否中文（中文选国内）
模型大小
硬件兼容性

数据准备
数据不同格式训练不同模型
重复数据删除否则增加训练时长

comfyui

本地运行大模型的一种方式

发表于 2025-09-21

本文仅介绍如何运行大模型
前提：有足够内存

云主机上运行

创建一个云主机
使用华为云主机https://developer.huaweicloud.com/space/home, 目前有180hours/年免费使用时长
登录后
工作台–配置一个云主机（需要实名认证，微信刷脸）– 点击进入桌面
注意这里进入的ubantu 系统，操作略有区别
安装Ollama
云主机界面右键”Open Terminal Here”，打开命令终端窗口
执行命令：
1
curl -fsSL https://dtse-mirrors.obs.cn-north-4.myhuaweicloud.com/case/0035/install.sh | sudo bash
创建DeepSeek模型
借助 Ollama 工具来部署 Deepseek 大模型，部署 deepseek-r1:1.5b 版本，执行命令：
1
ollama run deepseek-r1:1.5b

如果需要查看其他模型可直接去官网
点击任一一个模型名称可进一步看到模型版本和大小

ollama 命令行可以看做是管理大模型的工具
[cli 文档] (https://docs.ollama.com/cli)
ollama run xxx 命令会自行判断模型是否存在，不存在则下载模型后运行
将指定模型运行后就可以和模型进行对话了

创建Agent智能角色

关闭上面terminal 之后，虚拟环境创建

打开云主机桌面CodeArts IDE for Python
点击新建工程，输入工程名称：Agent，点击创建
左下角选择：管理->设置
在搜索框中输入：activate，找到图中所示，在“Python:Terminal:Activate Environment”选项前方框中点击勾选，保证在终端运行时展示当前指定python 环境
打开终端，前方出现（venv）字样表示创建成功，如果没有请点击“删除”图标关闭终端重新打开一下

升级pip,下载Autogen框架,openai借口

1
2
3

pip install --upgrade pip -i https://mirrors.tuna.tsinghua.edu.cn/pypi/web/simple
pip install pyautogen -i https://mirrors.tuna.tsinghua.edu.cn/pypi/web/simple
pip install openai -i https://mirrors.tuna.tsinghua.edu.cn/pypi/web/simple

新建运行python文件用来装运行代码，配置文件指定模型相关配置供代码使用，提示词文件注入上下文，也是供代码使用

demo.py:

from autogen import UserProxyAgent, ConversableAgent, config_list_from_json
from openai import models
def main():
    with open(r'提示词.txt', encoding='utf-8') as f:
        jiudan = f.readlines()
        jiudan = str(jiudan)
    # Load LLM inference endpoints from an env variable or a file
    # 从文件或者环境变量中加载LLM的推理端点
    config_list = config_list_from_json(env_or_file="OAI_CONFIG_LIST.json")
    #利用LLM来创建一个Assistant Agent,这个LLM是在上面的配置文件里指定的。
    assistant = ConversableAgent("agent", llm_config={"config_list": config_list},system_message=jiudan)
    #创建一个用户代理，这个用户代理就代表你，你可以随时加入对话中
    user_proxy = UserProxyAgent("user", code_execution_config=False)

    #让assistant来开始这个对话，如果用户输入 exit，那么对话就会被终止。
    assistant.initiate_chat(user_proxy, message="欢迎来到私人健康助手")
if __name__ == "__main__":
    main()

OAI_CONFIG_LIST.json:

[
  {
    "model": "deepseek-r1:1.5b",
    "api_key": "testkey",
    "base_url": "http://localhost:11434/v1"
  }
]

提示词txt:

# 角色
你是以为热情的医疗顾问，擅长根据患者的症状推荐合适的资料方案，还能与患者畅谈分享健康知识。让咨询过程充满关怀和专业。
## 技能
### 技能1：推荐治疗方案，直接给出患者一种治疗方案，如果不合适再询问患者具体症状。
1.当患者表明症状时，从已知治疗方案中挑选合适的进行推荐。
2.若不清楚患者症状时，可询问患者具体症状、病史等信息。
3.专注药物治疗推荐：
    -对于患者提出的非药物治疗问题（如手术或物理治疗），清始终引导回药物治疗推荐。
    -对于和药物治疗无关的问题，请忽略，请始终引导回药物治疗推荐。
只能出现这里面的治疗方案[
    {
    "名称": "酚麻美敏片",
    "主要成分": "对乙酰氨基酚、盐酸伪麻黄碱、氢溴酸右美沙芬、马来酸氯苯那敏",
    "适应症": "用于普通感冒或流行性感冒引起的发热、头痛、四肢酸痛、打喷嚏、流鼻涕、鼻塞、咳嗽、咽痛等症状",
    "剂量": "成人和12岁以上儿童，一次1 - 2片，每6小时服1次，24小时内不超过4次",
    "副作用": "有时有轻度头晕、乏力、恶心、上腹不适、口干、食欲缺乏和皮疹等，可自行恢复",
    "注意事项": "用药3 - 7天，症状未缓解，请咨询医师或药师；服用本品期间不得饮酒或含有酒精的饮料；不能同时服用与本品成份相似的其他抗感冒药等",
    "适用人群": "成人和12岁以上儿童",
    "使用时长": "一般不超过7天",
    "颜色": "薄膜衣片，除去包衣后显白色或类白色"
},
{
    "名称": "氢溴酸右美沙芬糖浆",
    "主要成分": "氢溴酸右美沙芬",
    "适应症": "用于干咳，包括上呼吸道感染（如感冒和咽炎）、支气管炎等引起的咳嗽",
    "剂量": "口服。成人一次10 - 20毫升，一日3 - 4次",
    "副作用": "可见头晕、头痛、嗜睡、易激动、嗳气、食欲缺乏、便秘、恶心、皮肤过敏等，但不影响疗效。停药后上述反应可自行消失",
    "注意事项": "用药7天，症状未缓解，请咨询医师；哮喘患者、痰多的患者、肝肾功能不全患者慎用；孕妇及哺乳期妇女慎用等",
    "适用人群": "成人",
    "使用时长": "一般不超过7天",
    "颜色": "棕黄色澄清液体"
},
{
    "名称": "氯雷他定片",
    "主要成分": "氯雷他定",
    "适应症": "用于缓解过敏性鼻炎有关的症状，如喷嚏、流涕、鼻痒、鼻塞以及眼部痒及烧灼感。口服药物后，鼻和眼部症状及体征得以迅速缓解。亦适用于缓解慢性荨麻疹、瘙痒性皮肤病及其他过敏性皮肤病的症状及体征",
    "剂量": "成人及12岁以上儿童：一日1次，一次1片（10毫克）；2 - 12岁儿童：体重＞30公斤：一日1次，一次1片（10毫克）；体重≤30公斤：一日1次，一次半片（5毫克）",
    "副作用": "常见不良反应有乏力、头痛、嗜睡、口干、胃肠道不适包括恶心、胃炎以及皮疹等。罕见不良反应有脱发、过敏反应、肝功能异常、心动过速及心悸等",
    "注意事项": "在做皮肤过敏试验前约48小时应中止使用本品；同时服用酮康唑、大环内酯类抗生素、西咪替丁、茶碱等药物，会提高氯雷他定在血浆中的浓度，应慎用；罕见不良反应有心动过速、心悸、肝酶升高、肝炎及胆红素增加等，应停药并就医等",
    "适用人群": "成人及2岁以上儿童",
    "使用时长": "根据病情而定，症状缓解后可停药，一般慢性疾病疗程较长",
    "颜色": "白色或类白色片"
},
{
    "名称": "健胃消食片",
    "主要成分": "太子参、陈皮、山药、麦芽（炒）、山楂",
    "适应症": "健胃消食。用于脾胃虚弱所致的食积，症见不思饮食、嗳腐酸臭、脘腹胀满；消化不良见上述证候者",
    "剂量": "口服，可以咀嚼。成人一次4 - 6片，儿童二岁至四岁一次2片，五岁至八岁一次3片，九岁至十四岁一次4片；一日3次",
    "副作用": "尚不明确",
    "注意事项": "饮食宜清淡，忌食辛辣、生冷、油腻食物；孕妇及哺乳期妇女慎用；儿童必须在成人监护下使用等",
    "适用人群": "各年龄段人群，儿童需在成人监护下使用",
    "使用时长": "症状缓解后即可停药",
    "颜色": "浅棕黄色的片；气微香，味微甜、酸"
},{
    "名称": "硝苯地平缓释片",
    "主要成分": "硝苯地平",
    "适应症": "各种类型的高血压及心绞痛",
    "剂量": "口服。一次10 - 20毫克，一日2次，极量，一次40毫克，一日0.12克",
    "副作用": "可能出现的副作用有外周水肿（外周水肿与剂量相关，服用120片/日时发生率为4.8%，服用240片/日则为12.8%）、头晕、头痛、脸红、心悸、乏力、恶心等",
    "注意事项": "低血压、严重主动脉瓣狭窄、肝肾功能不全患者慎用；用药期间定期监测血压和心率；避免突然停药，以免引起血压反跳等",
    "适用人群": "高血压及心绞痛患者",
    "使用时长": "需长期服用，根据血压控制情况调整剂量",
    "颜色": "黄色薄膜衣片除去薄膜衣后显黄色"
}
]

保证三者在同一个文件夹下

然后打开CodeArts IDE for Python 终端
执行python demo.py
大模型就可以更准确的回答和提示词相关的问题了

本地运行

在实际自己的电脑笔记本上运行

步骤与上面类似，工具略有不同

安装ollama, 帮忙下载模型
下载完成后，可直接在笔记本终端跑ollama 命令同上
点击客户端也可以直接提问，前提也是已经有可用大模型model，没有的话，直接选择也是会先下载再运行，
执行下
ollama run xxx
下载一个模型，供open webUI使用
升级python 到3.11以上，装 open webUI
1
pip install open-webui

安装完后运行

1	open-webui serve

访问 http://localhost:8080/
如无法访问，等一会儿再刷新

模型配置
可避免建上面云主机的文件，直接定制自己的agent
在open webUI的界面中选择【工作空间】- 添加模型（版本不同，操作界面不同，进入下面页面即可）

后记：没必要在本地部署，尤其本地，内存消耗巨大，尤其是模型本身体积巨大

其他问题记录

决策分析式AI 与生成式AI 区别
二者本质的不同是从底层开始演绎的算法不同，而且最终生成的产物应用阈也不同
决策分析式AI
通过对历史数据的学习，辅助我们进行一些决策，应用阈更加垂类，
与数据挖掘，机器学习是一个时代的概念，相关算法设计决策树，KNN, 集成算法，朴素贝叶斯，关联规则，推荐算法
生成式AI
通过对大量历史数据进行学习，帮助我们进行内容创作，这些内容通常都是全新的，应用阈更加宽泛，通用，涉及的算法有卷积神经网络等
模型压缩方法？？？？
量化：模型参数的量化，降低数据精度，将模型参数从浮点数转换为整数，从而减少存储空间和计算量。
结构化剪枝：剔除模型中无意义或冗余的注意头
蒸馏：比如使用LoRA 算法进行降秩处理
语义理解
大模型实现语义理解的关键目前是使用transform机制，transform机制是多个并行执行的
装包问题
pyenv install 3.12.0 遇到 ModuleNotFoundError: No module named ‘_lzma’
执行下面命令后再次安装
1
brew install xz

pip install open-webui 安装过程始终联网，这个耗时长，如果中断了，再执行一次就可以

执行open-webui serve 报错 ModuleNotFoundError: No module named ‘itsdangerous’
先装包再次执行

1	pip install itsdangerous

系统思考

发表于 2025-06-22

系统

由相互依赖，具有特殊目的多个部分组成的整体

多个部分
相互依赖
具有特殊目的

系统思考角度

深度思考: 从现象到本质
全局思考: 从局部思考到整体思考
动态思考: 理解每个人每件事之间的关系都是动态变化的

因果回路图

变量

系统结构里面的因素，会随时间而变化的名词

链路: 变量之间的因果关系，也称因果链路，分为正相关链路(A+B+)和负相关链路(A+B-)

回路

通过多条链路形成一个闭合的圈，分为增强回路和平衡回路

增强回路

所有链路都是正相关的
Reinforcing loop，一个回路中的变量增加或减少，会影响这个回路中的所有链路持续增加或减少，发展的趋势不受控制，
常见的类比说法比如“恶性循环”、“强者恒强”等等就是增强回路导致的；

平衡回路

有正相关也有负相关
Balance loop，一个回路中的变量增加或减少受到系统中其他变量的反向影响，使得这个系统中的变量在长期的维度会表现出一种保持平衡的状态，
比如最常见的例子是，猪肉如果大幅度涨价，就会有更多的人加入到养猪的行业，第二年的猪肉就会因为供应充足而降价，最终长期看价格会维持在一个平衡的状态。

延时

一个变量的变化影响另一个变量并不一定是马上生效的，他们之间的关系有可能存在时延
在链路之间增加一个“||”的符号代表这两个变量之间的因果关系存在时延

时延在工作中最典型的例子比如：
招聘对项目人力缺口的影响、代码单元测试对产品质量的影响、学习对于工作能力的影响等等。
对时延的感知也是帮助理解系统复杂性的重点之一。

系统思考的五个基础模型

通过执行不同的决策最终导致平衡或者增强回路的出现，没有解决实际问题的模型
可以更好的帮助了解分析定位系统中出现问题的因果流程

饮鸩止渴

B：走捷径为了加速
R：有故障会拖后腿

饮鸩止渴”描述了我们是怎么在进度的压力下一次又一次的放弃了自己的坚持，因为链路上的延迟，让我们心存侥幸，最后使得我们的系统背负了沉重的技术债的。
左侧半圈虽然是个平衡回路，但是整个外圈回路是个增强回路【负负得正】

舍本逐末

B1：走捷径为了加速【末】
B2：优化架构做加速【本】
R：陷入架构泥潭无法脱身【因延时】

“舍本逐末”描述了短期表面方案和长期根本方案之间的冲突，因为增强回路的存在，使得我们不能对“架构优化”这个根本的方案提高优先级，最终上瘾于短期表面方案

目标侵蚀

B1：通过行动推进【本】
B2：通过修改目标推进【走捷径】
R：完成目标的动力被侵蚀【对实际进展产生负影响】

“目标侵蚀”描述了我们怎么在目标完成的压力下，放弃了做正确的事，而是通过直接降低目标来达成目标的。真实的“加速”措施通常需要更长的时间才能见效。正是这个延迟，使得我们逐步转向上面的平衡回路，需求延期和下调目标成为一种习惯

成长上限

R：业务增长飞轮
B：业务增长天花板【人口规模是一定的，不可能一直增】

“成长上限”描述了一个增强回路不可能独自持续下去，在一个更大的维度，一定会有另一个因素（或平衡回路）对它进行限制，这个就是成长上限。

公地悲剧

R：投放即增长
B：体验下降，用户用脚投票【最外面一圈是个平衡回路】

“公地悲剧”描述了对于大家共享的有限资源（push渠道），每个个体（业务单元）都想自己利益最大化。
使用者越多，越消耗用户对平台体验的信任。
随着push总量迅速增加，遭遇用户容忍瓶颈时，消费者会感到不可容忍，用脚投票。

更复杂的例子

回归思考的心智模型

怎么找到合适的“变量”

从目标或问题出发
- 明确系统目标：首先确定系统的核心目标（如 “提高产品销量”“降低成本”），围绕目标反推影响它的关键因素。
- 聚焦核心问题：从具体问题出发，拆解问题背后的驱动因素。例如，“员工流失率高” 可能涉及 “薪资水平”“晋升机会”“工作压力” 等变量。
逐层分解系统
- MECE 原则（相互独立、完全穷尽）：将复杂系统分解为互不重叠且覆盖全面的子系统。例如，分析企业运营时，可分解为 “市场”“生产”“财务”“人力资源” 等模块，再分别提取变量。
- 5Why 分析法：通过连续追问 “为什么”，挖掘深层变量。
从参与者视角切入
- 列出所有利益相关者：识别与系统相关的个体或群体（如PM、开发、测试、DA等），分析他们的行为如何影响系统。
- 寻找交叉影响：关注不同利益相关者之间的互动。例如，“排期压力” 可能影响 “软件质量”，进而影响 “用户体验”。

系统思考的心智模型

对于复杂问题的思考是有层次的，从最表面的事件（正在发生什么），到事件背后的规律（发展趋势是什么），再到这个问题的结构模式（解释趋势背后的原因），再到价值观（驱动这个模式的理念），层层递进。

在画完自己的业务系统因果回路图之后，再结合这个心智模型，思考自己的思考在哪个层次，是否可以有机会再下钻到更深的层次。

值得注意的地方，“系统思考”只是一个工具，不同的人，面对同样一个系统，因为了解的信息多少不同，关注的问题角度不同，对系统发展方向的期待不同，都会导致画出来的因果回路图有所不同。

所以，“系统思考”就是一个帮助你不断的通过zoom out、zoom in 来完整的、体系的看待复杂问题的工具，通过使用这个工具的过程，帮助更好的思考和理解你面对的复杂问题。

前端组件如何注册到后端agent服务中的

三类组件注册流程总览

多样的UI界面是谁渲染的

UI 渲染总览流程

controlled Generative UI 渲染逻辑

非controlled Generative UI 渲染逻辑

a2UI

mcp 渲染逻辑

open Generative UI 渲染逻辑

CopilotRuntime

AGUI 协议

abstractAgent

httpAgent

runHttpRequest

transformHttpEventStream

parseSSEStream

parseProtoStream

为什么默认 copilotKit 实例不同动作 可以发送不同http path 请求

基本使用

返回结果

参数

多模态输入

流式输出

提示词

prompt cache

tool

Generative UI

AG-UI 协议

区别其他协议

区别 A2UI

CopilotKit 框架

generative ui

hooks

多agent 切换

后端启动多服务

前端注册多个agent

前端指定agent

前端调用CopilotKit agent

Controlled Generative UI 全控式

useComponent 注册组件

Deccontrolled Generative UI 声明式

什么是声明式生成式 UI？

为什么使用声明式生成式 UI？

开启a2ui 组件 进行组件组装渲染

dynamic schema UI

注册

fixed schema UI

将ui schema 注册到后端agent 上

Dynamic vs Fixed: When to use which

Open Generative UI 开放式

全栈状态共享

Sequence to Sequence Models

集束搜索算法

优化

问题分析

bleu 算法

注意力机制

应用

transformer

自注意力机制

多头注意力机制

transform 网络

word representation 词汇表征

词嵌入应用

实体识别 ==> 迁移学习

类比推理

词嵌入矩阵

使用word embedding 进行模型训练

优化

文生视频

商品评论分析

推荐算法1

推荐算法2

代码的方式

使用 knime

业务场景分类

任务类型

硬件选择

计算资源

模型选择/模型类型

为什么默认 copilotKit 实例不同动作可以发送不同http path 请求

开启a2ui 组件进行组件组装渲染