from google.adk.agents import Agent
from google.adk.tools import FunctionTool
from google.adk.runners import Runner
from google.adk.sessions import InMemorySessionService
from pymilvus import connections,Collection,DataType
# ---------- Session Service ----------
session_service = InMemorySessionService # 每个会话独立
# ---------- Memory Service ----------
connections.connect
memory_collection = Collection
def recall_memory -> str:
query_emb = genai.embed_content(
model='models/text-embedding-004'。content=query)
results = memory_collection.search(
data=,anns_field='embedding',param={'metric_type': 'COSINE','params': {'nprobe': 10}},limit=top_k,expr=f'user_id == "{user_id}"',output_fields=)
formatted = }
{hit.entity}..."
for i,hit in enumerate]
return "
".join
recall_memory_tool = FunctionTool(
name='recall_memory',description='从长期记忆检索相关案例',function=recall_memory)
support_agent = Agent(
model='gemini-flash-lite',name='support_agent',description='技术支持专家',instruction='''
你是技术支持专家。说到流程，1️⃣ 调用 recall_memory 检索历史案例；2️⃣ 基于检索结果回答；3️⃣ 完成后询问使用者是否解决；4️⃣ 若确认，则调用 store_memory 存储新案例。''',tools=)
runner = Runner(agent=support_agent,app_name='tech_support'。session_service=session_service)

┌─────────────────────────────────────────────────────────┐
│ Managed Agents Architecture │
│ │
│ ┌──────────────┐ ┌──────────────┐ ┌────────────┐|
│ │ Session │ │ Harness │ │ Sandbox │|
│ │ | │ | │ |
│ ├──────────────┤ ├──────────────┤ ├────────────┤|
│ • 持久事件日志 • • 工具调用决策 • • 代码执行 |
│ • 上下文外存储 • • 上下文管理 • • 文件编辑 |
│ • 可恢复/可查询 • • 错误恢复 • • 权限隔离 |
│ |
│ Credential Proxy / Vault |
└─────────────────────────────────────────────────────────┘

# Session 接口 — 持久 Append‑Only 日志
def getEvents -> List: ...
def emitEvent -> None: ...
def wake -> HarnessState: ...
# Harness 接口 — 编排 & 控制面
def execute -> str # 标准化工具调用
def provision -> Sandbox # 动态创建执行环境
# Sandbox 接口 — 执行环境
def execute -> str
def checkpoint -> Snapshot # 保存执行状态
def restore -> None # 恢复执行状态