用 Omnigent 构建策略管控的多智能体金融研究工作流

In this tutorial, we build and execute a multi-agent workflow with Omnigent using a reliable, isolated Python environment created with uv. We configure a financial research lead agent that retrieves a live USD-to-EUR exchange rate from an external API, prepares a concise client-ready summary, and delegates its draft to a dedicated text-auditing sub-agent for clarity and length validation. We define reusable Python functions as callable agent tools, describe the complete agent structure in YAML, and use the Claude Agent SDK as the execution harness. We also manage the Anthropic API key securely through environment variables, apply non-interactive policies that limit tool calls and control session costs, and run the workflow directly from Colab without requiring Node.js, tmux, or an interactive terminal. Through this implementation, we explore how Omnigent combines agents, tools, delegation, live data access, and governance within a single configurable system.

import os, sys, subprocess, textwrap, pathlib, getpass
def sh(cmd, **kw):
   """Run a command, and on failure show the ACTUAL error, not just a code."""
   print("$", " ".join(map(str, cmd)))
   p = subprocess.run(cmd, text=True, capture_output=True, **kw)
   if p.returncode != 0:
       print(p.stdout or "", p.stderr or "", sep="\n")
       raise RuntimeError(f"Command failed ({p.returncode}): {' '.join(map(str, cmd))}")
   return p
WORKDIR = pathlib.Path("/content/omnigent_tutorial")
WORKDIR.mkdir(parents=True, exist_ok=True)
VENV = WORKDIR / ".venv"
subprocess.run([sys.executable, "-m", "pip", "install", "-q", "uv"], check=True)
if not (VENV / "bin" / "python").exists():
   sh(["uv", "venv", "--python", "3.12", str(VENV)])
PY = str(VENV / "bin" / "python")
sh(["uv", "pip", "install", "--python", PY, "-q", "omnigent", "requests"])
OMNI = str(VENV / "bin" / "omnigent")
print("\n✅", subprocess.run([OMNI, "--version"], capture_output=True, text=True).stdout.strip())

We import the required Python modules and define a helper function that executes shell commands while displaying detailed error information when a command fails. We create a dedicated working directory and use uv to build an isolated Python 3.12 virtual environment that avoids Colab’s ensurepip limitation. We then install Omnigent and Requests inside the environment, locate the Omnigent CLI executable, and verify the installation by printing its version.

if not os.environ.get("ANTHROPIC_API_KEY"):
   os.environ["ANTHROPIC_API_KEY"] = getpass.getpass("Anthropic API key: ")
env = os.environ.copy()
env["OMNIGENT_NO_UPDATE_CHECK"] = "1"

We securely collect the Anthropic API key only when it is not already available in the notebook environment. We store the credential in the current process environment so that Omnigent can detect it without writing sensitive information to a file. We also create a separate environment configuration for the subprocess and turn off Omnigent’s automatic update check during execution.

(WORKDIR / "agent_tools.py").write_text(textwrap.dedent('''
   """Local tools exposed to the Omnigent agents in this tutorial."""
   import requests
   def get_exchange_rate(base_currency: str, target_currency: str) -> dict:
       """Look up the latest FX rate between two ISO-4217 currency codes."""
       r = requests.get(
           "https://api.frankfurter.app/latest",
           params={"from": base_currency.upper(), "to": target_currency.upper()},
           timeout=10,
       )
       r.raise_for_status()
       data = r.json()
       return {
           "base": base_currency.upper(),
           "target": target_currency.upper(),
           "rate": data["rates"][target_currency.upper()],
           "date": data["date"],
       }
   def word_count(text: str) -> int:
       """Count the words in a piece of text."""
       return len(text.split())
'''))

We generate a Python module containing the local functions that Omnigent exposes as callable tools to the agents. We define a live exchange-rate tool that sends a request to the Frankfurter API and returns the latest rate, currency codes, and applicable date. We also implement a simple word-count tool that allows the auditing sub-agent to measure the length of the financial summary.

(WORKDIR / "fx_research_lead.yaml").write_text(textwrap.dedent('''
   name: fx_research_lead
   prompt: |
     You are a financial research lead. For any question about currency
     movements: call get_exchange_rate to fetch the live rate, then hand
     your draft summary to the text_auditor sub-agent for a clarity and
     length check before giving your final answer to the user.
   executor:
     harness: claude-sdk
   tools:
     get_exchange_rate:
       type: function
       callable: agent_tools.get_exchange_rate
     text_auditor:
       type: agent
       prompt: |
         You audit short pieces of financial writing. Call word_count to
         report its length, flag any unexplained jargon, and suggest one
         concrete clarity improvement.
       tools:
         word_count:
           type: function
           callable: agent_tools.word_count
   policies:
     cap_calls:
       type: function
       handler: omnigent.policies.builtins.safety.max_tool_calls_per_session
       factory_params:
         limit: 20
     budget:
       type: function
       handler: omnigent.policies.builtins.cost.cost_budget
       factory_params:
         max_cost_usd: 1.00
'''))

We define the complete multi-agent architecture through a YAML configuration file. We configure the financial research lead, connect it to the exchange-rate tool, and add a text-auditing sub-agent that evaluates the draft using the word-count function. We also apply hard governance policies that restrict the number of tool calls and limit the maximum API cost for the session.

env["PYTHONPATH"] = str(WORKDIR)
question = (
   "What is the current USD to EUR exchange rate? Give me a two-sentence "
   "summary I could paste into a client note."
)
result = subprocess.run(
   [OMNI, "run", str(WORKDIR / "fx_research_lead.yaml"), "-p", question, "--no-session"],
   cwd=WORKDIR, env=env, stdin=subprocess.DEVNULL,
   capture_output=True, text=True, timeout=300,
)
print("\n" + "=" * 70)
print(result.stdout.strip() or "(no stdout)")
if result.returncode != 0 or "error" in result.stdout.lower():
   print("-" * 70)
   print("stderr:", result.stderr[-2000:])
   print(f"\nDebug: check ~/.omnigent/logs/runner/ , or rerun with:\n"
         f"  !{OMNI} --debug --log-to-stderr run {WORKDIR/'fx_research_lead.yaml'} -p \"...\" --no-session")
print("=" * 70)
print(f"""
Next steps:
 • Explore the CLI:  !{OMNI} run --help
 • Bundled demo agents:
       !{OMNI} polly -p "review this repo" --no-session
       !{OMNI} debby -p "brainstorm 3 names for a coffee shop" --no-session
 • YAML schema:  https://github.com/omnigent-ai/omnigent/blob/main/docs/AGENT_YAML_SPEC.md
 • Policies:     https://github.com/omnigent-ai/omnigent/blob/main/docs/POLICIES.md
""")

We add the tutorial directory to PYTHONPATH, define the currency-related question, and execute the Omnigent agent through a non-interactive subprocess. We capture the generated response, display diagnostic output when execution fails, and provide a debug command for examining runner issues. We finish by printing useful next steps for exploring Omnigent’s CLI, bundled agents, YAML specification, and policy documentation.

In conclusion, we created a practical Omnigent multi-agent application that integrates live financial data retrieval, hierarchical agent delegation, automated writing assessment, and policy-based execution controls. We used uv to solve Colab’s ensurepip limitation and maintain a separate Python 3.12 environment without modifying the notebook’s system interpreter. We exposed local Python functions as agent-accessible tools, defined the agent and sub-agent behavior through a readable YAML configuration, and enforced hard limits on tool usage and API spending. We also executed the workflow non-interactively, captured both standard output and diagnostic errors, and established a structure that supports easy model changes without altering the underlying agent logic. We now have a reusable foundation for developing more sophisticated, secure, cost-controlled, and tool-enabled multi-agent systems for financial research and other real-world applications in Google Colab.