CrewAI 基础 07-Knowledge

发表于 更新于 分类于 ， ， 阅读次数： 本文字数： 6.9k 阅读时长 ≈ 6 分钟

在 crewai 中，使用记忆系统可以给 Agent 或者 Crew 可以赋予一个角色、团队临时性或者经历过的知识，对于团队外部即时的、更加庞大的外部知识，将无法使用，因此给 Agent、Crew 引入 knowledge 的概念，以供 Agent、Crew 在执行 Task 时，查看这些知识

相比较 memory, knowledge 有以下优势：

  • 类似 RAG，可以使用领域内特定的知识
  • 即时的知识

让我们深入 crewai，看看 knowledge 的构建、使用

classDiagram 
    BaseKnowledgeSource <|-- BaseFileKnowledgeSource
    BaseKnowledgeSource <|-- StringKnowledgeSource
    BaseFileKnowledgeSource <|-- ExcelKnowledgeSource

    class BaseKnowledgeSource{
        chunk_size:int=4000
        chunk_overlap: int = 200
        chunks: List[str]
        chunk_embeddings: List[np.ndarray]
        model_config
        storage: KnowledgeStorage
        metadata: Dict[str, Any]
        collection_name: Optional[str] 
    
        load_content(): Dict[Path, str]
        add()
        get_embeddings()
        _chunk_text(text: str): List[str]
        save_documents(metadata)
    }
    class BaseFileKnowledgeSource{
        file_path: 
        content: Dict[Path, str]
        storage: KnowledgeStorage
        load_content(): Dict[Path, str]
        save_documents(metadata)
    }
    class StringKnowledgeSource{
        content: str
        collection_name: Optional[str]
        add()
        _chunk_text(text: str): List[str]        
    }
    class ExcelKnowledgeSource{
        load_content()
        add()
        _chunk_text()
    }

从代码中可以看出，CrewAI 支持两大类知识源，即：1）文本源，如字符串、文本文件、PDF 文件、Markdown 文件；2）结构化数据源，如 csv、excle、json 等

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
from crewai import Agent,Task,Crew,Process,LLM

llm=LLM(
    model="ollama_chat/qwen2.5:latest", 
    base_url="http://localhost:11434")

embedder={
    "provider": "ollama",
    "config": {
        "model": 'quentinz/bge-large-zh-v1.5:latest',
        "base_url": 'http://localhost:11434'
    }
}

from crewai.knowledge.source.string_knowledge_source import StringKnowledgeSource
string_source=StringKnowledgeSource(
    content="小伍，正在学习crewai的knowledge部分知识"
)

agent=Agent(
    role="客服",
    goal="以礼貌友好的态度回答用户问题",
    backstory="你是一个经验丰富的客服，总能收获用户的认可",
    verbose=True,
    allow_delegation=False,
    llm=llm
)

task=Task(
    description="回答用户问题：{question}",
    expected_output="回答用户问题",
    agent=agent
)

crew=Crew(
    agents=[agent],
    tasks=[task],
    verbose=True,
    manager_llm=llm,
    planning_llm=llm,
    process=Process.sequential,
    knowledge_sources=[string_source],
    embedder=embedder
)

result=crew.kickoff(inputs={"question":"小伍在做什么？"})

CrewAI基础07-Knowledge-20250107111135

不仅可以给团队设置外部知识，还可以给 agent 设置外部知识

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
string_source=StringKnowledgeSource(
    content="小伍，你希望成为一个怎样的人？控制自己能控制的，掌握自己不能控制，更好的物质条件，在每个抉择的十字路口，有更多的选择",
    metadata={"类别":"小伍的期待"}
)

agent=Agent(
    role="customer",  # 注意：这里创建向量数据库使用了该字符串，使用中文时，无法创建
    goal="以礼貌友好的态度回答用户问题",
    backstory="你是一个经验丰富的客服，总能收获用户的认可",
    verbose=True,
    allow_delegation=False,
    llm=llm,
    knowledge_sources=[string_source],
    embedder_config=embedder
)

task=Task(
    description="回答用户问题：{question}",
    expected_output="回答用户问题",
    agent=agent
)

crew=Crew(
    agents=[agent],
    tasks=[task],
    verbose=True
)

result=crew.kickoff(inputs={"question":"小伍对未来有什么期待？"})

CrewAI基础07-Knowledge-20250107111112

除了以上使用 creawai 官方的知识源外，还可以自定义知识源，以下例子我们将百度百科的相关词条整理为外部知识源

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
# 利用https://baike.deno.dev/获取百度百科查询地址
import re
import requests
from bs4 import BeautifulSoup

def get_html(url):
    headers = {
        'User-Agent': 'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/91.0.4472.124 Safari/537.36'
    }
    response = requests.get(url, headers=headers, stream=True, timeout=(5,5))
    
    # 检查响应状态码
    if response.status_code == 200:
        return response
    else:
        print(f"请求失败，状态码：{response.status_code}")
        return None

def build_md_from_html(html):
    soup = BeautifulSoup(html, 'html.parser')

    # 获取summary部分
    md_summary='# 概览\n'
    summary_div= soup.find_all('div', class_='J-summary')  # 替换为实际的数据选择器
    summary_paragraphs = summary_div[0].find_all('div', {'data-tag': 'paragraph'})
    for paragraph in summary_paragraphs:
        text = paragraph.get_text(strip=True)
        text=re.sub(r'\[[1-9][0-9]*\]','',text)
        md_summary+=text+'\n\n'
    
    # 获取内容部分
    md_content=''
    content_div = soup.find_all('div', class_='J-lemma-content')  # 替换为实际的数据选择器
    paragraphs = content_div[0].find_all('div', {'data-tag': ['paragraph','header']})
    for paragraph in paragraphs:
        data_tag=paragraph['data-tag']
        text = paragraph.get_text(strip=True)

        if data_tag=='header':
            text=text.replace('播报编辑','')
            data_level=int(paragraph['data-level'])
            text='#'*data_level+' '+text+'\n'
        else:
            text=re.sub(r'\[[1-9][0-9]*\]','',text)
            text+='\n\n'
        
        md_content+=text

    # 将结果写入文件
    return md_summary,md_content

from crewai import Agent,Task,Crew,Process,LLM
from crewai.knowledge.source.base_file_knowledge_source import BaseKnowledgeSource

from typing import Dict,Any,List
from pydantic import BaseModel,Field

class BaiKeKonwledgeSource(BaseKnowledgeSource):
    """从百度百科构建知识源"""
    keywords:List[str]=Field(description="用于百科查询的关键词")

    def validate_content(self) -> Dict[Any, str]:
        """获取百度百科知识内容"""
        # 收集百科返回文本
        knowledge_list=[]
        for keyword in self.keywords:
            print('fetch content by keyword',keyword)
            url = f"https://baike.deno.dev/item/{keyword}"
            response = get_html(url).json()
            baike_url=response['data']['link']
            print('baike url:',baike_url)

            md_summary,md_content=build_md_from_html(get_html(baike_url).text)
            article=f"""
                Title:{keyword}
                URL:{baike_url}
                Summary:{md_summary}
                Content:{md_content}
                """
            knowledge_list.append(article)
        
        # 聚合所有文本
        knowledge_content='||'.join(knowledge_list)

        return {'baike':knowledge_content}


    def add(self) -> None:
        """处理和存储百科内容"""
        content = self.validate_content()
        for _, text in content.items():
            chunks = self._chunk_text(text)
            self.chunks.extend(chunks)

        self._save_documents()

keywords=["自动驾驶系统", "北京市自动驾驶汽车条例","自动驾驶路测牌照"]
baike_knowledge_source=BaiKeKonwledgeSource(
    keywords=keywords,
    chunk_size=512,
    chunk_overlap=30)

# 这段代码有个问题，当把知识赋予agent时，其role必须是英文，导致输出必定英文，目前将知识赋予团队解决
auto_driving_expert=Agent(
    role="自动驾驶专家",
    goal="普及自动驾驶知识，向广大群众推广自动驾驶",
    backstory="在头部汽车制造公司担任市场部总监多年，熟悉自动驾驶的原理，风险",
    llm=llm
)

discuss_task=Task(
    description="发表一篇关于开放L3级别自动驾驶演讲，讲明白L3级别自动驾驶的定义，开展测试的注意事项，违法交通规则的处理措施等",
    agent=auto_driving_expert,
    expected_output="以中文输出，每个要点输出一段文字，不少于300字，输出为markdow格式"
)

crew=Crew(
    agents=[auto_driving_expert],
    tasks=[discuss_task],
    verbose=True,
    process=Process.sequential,
    language='Chinese',
    knowledge_sources=[baike_knowledge_source],
    embedder=embedder
)

result=crew.kickoff()

输出：

CrewAI基础07-Knowledge-20250107111030
​